|
| 1 | +import inspect |
1 | 2 | from typing import Any, Dict, List, Optional, Union |
2 | 3 |
|
3 | 4 | from langchain_core.language_models.chat_models import BaseChatModel |
@@ -82,6 +83,41 @@ def create_langchain_model(ai_config: AIConfigKind) -> BaseChatModel: |
82 | 83 | ) |
83 | 84 |
|
84 | 85 |
|
| 86 | +def _iter_valid_tools( |
| 87 | + tool_definitions: List[Dict[str, Any]], |
| 88 | + tool_registry: ToolRegistry, |
| 89 | +) -> List[tuple]: |
| 90 | + """ |
| 91 | + Filter LD tool definitions against a registry, returning (name, td) pairs for each |
| 92 | + valid function tool that has a callable implementation. Built-in provider tools and |
| 93 | + tools missing from the registry are skipped with a warning. |
| 94 | + """ |
| 95 | + valid = [] |
| 96 | + for td in tool_definitions: |
| 97 | + if not isinstance(td, dict): |
| 98 | + continue |
| 99 | + |
| 100 | + tool_type = td.get('type') |
| 101 | + if tool_type and tool_type != 'function': |
| 102 | + log.warning( |
| 103 | + f"Built-in tool '{tool_type}' is not reliably supported via LangChain and will be skipped. " |
| 104 | + "Use a provider-specific runner to use built-in provider tools." |
| 105 | + ) |
| 106 | + continue |
| 107 | + |
| 108 | + name = td.get('name') |
| 109 | + if not name: |
| 110 | + continue |
| 111 | + |
| 112 | + if name not in tool_registry: |
| 113 | + log.warning(f"Tool '{name}' is defined in the AI config but was not found in the tool registry; skipping.") |
| 114 | + continue |
| 115 | + |
| 116 | + valid.append((name, td)) |
| 117 | + |
| 118 | + return valid |
| 119 | + |
| 120 | + |
85 | 121 | def build_tools(ai_config: AIConfigKind, tool_registry: ToolRegistry) -> List[Any]: |
86 | 122 | """ |
87 | 123 | Return callables from the registry for each tool defined in the AI config. |
@@ -114,6 +150,39 @@ def build_tools(ai_config: AIConfigKind, tool_registry: ToolRegistry) -> List[An |
114 | 150 | return tools |
115 | 151 |
|
116 | 152 |
|
| 153 | +def build_structured_tools(ai_config: AIConfigKind, tool_registry: ToolRegistry) -> List[Any]: |
| 154 | + """ |
| 155 | + Build a list of LangChain StructuredTool instances from LD tool definitions and a registry. |
| 156 | +
|
| 157 | + Tools found in the registry are wrapped as StructuredTool using the LD config key as the |
| 158 | + tool name so the model's tool calls match ToolNode lookup. Async callables use ``coroutine=`` |
| 159 | + so LangGraph invokes them correctly. Built-in provider tools and tools missing from the |
| 160 | + registry are skipped with a warning. |
| 161 | +
|
| 162 | + :param ai_config: The LaunchDarkly AI configuration |
| 163 | + :param tool_registry: Registry mapping tool names to callable implementations |
| 164 | + :return: List of StructuredTool instances ready to pass to langchain.agents.create_agent |
| 165 | + """ |
| 166 | + from langchain_core.tools import StructuredTool |
| 167 | + |
| 168 | + config_dict = ai_config.to_dict() |
| 169 | + model_dict = config_dict.get('model') or {} |
| 170 | + parameters = dict(model_dict.get('parameters') or {}) |
| 171 | + tool_definitions = parameters.pop('tools', []) or [] |
| 172 | + |
| 173 | + tools = [] |
| 174 | + for name, td in _iter_valid_tools(tool_definitions, tool_registry): |
| 175 | + fn = tool_registry[name] |
| 176 | + raw_desc = td.get('description') if isinstance(td.get('description'), str) else '' |
| 177 | + description = raw_desc.strip() or (getattr(fn, '__doc__', None) or '').strip() or f'Tool {name}' |
| 178 | + if inspect.iscoroutinefunction(fn): |
| 179 | + tool = StructuredTool.from_function(coroutine=fn, name=name, description=description) |
| 180 | + else: |
| 181 | + tool = StructuredTool.from_function(fn, name=name, description=description) |
| 182 | + tools.append(tool) |
| 183 | + return tools |
| 184 | + |
| 185 | + |
117 | 186 | def get_ai_usage_from_response(response: Any) -> Optional[TokenUsage]: |
118 | 187 | """ |
119 | 188 | Extract token usage from a LangChain response. |
|
0 commit comments