Translate LangChain ``ModelRequest`` payloads for request intercepts.
| 53 | |
| 54 | |
| 55 | class LangChainCodec(LlmCodec): |
| 56 | """Translate LangChain ``ModelRequest`` payloads for request intercepts.""" |
| 57 | |
| 58 | @classmethod |
| 59 | def _langchain_tool_calls_to_annotated(cls, tool_calls: list[Any]) -> list[dict[str, Any]]: |
| 60 | annotated_tool_calls = [] |
| 61 | for tool_call in tool_calls: |
| 62 | args = tool_call["args"] |
| 63 | arguments = args if isinstance(args, str) else json.dumps(args) |
| 64 | annotated_tool_calls.append( |
| 65 | { |
| 66 | "id": tool_call.get("id") or "", |
| 67 | "type": "function", |
| 68 | "function": { |
| 69 | "name": tool_call["name"], |
| 70 | "arguments": arguments, |
| 71 | }, |
| 72 | } |
| 73 | ) |
| 74 | |
| 75 | return annotated_tool_calls |
| 76 | |
| 77 | @classmethod |
| 78 | def _annotated_tool_calls_to_langchain(cls, tool_calls: Any) -> list[dict[str, Any]] | None: |
| 79 | if not isinstance(tool_calls, list) or not tool_calls: |
| 80 | return None |
| 81 | |
| 82 | langchain_tool_calls = [] |
| 83 | for tool_call in tool_calls: |
| 84 | if not isinstance(tool_call, dict): |
| 85 | continue |
| 86 | function = tool_call.get("function") |
| 87 | if isinstance(function, dict): |
| 88 | name = str(function.get("name") or "") |
| 89 | arguments = function.get("arguments", {}) |
| 90 | else: |
| 91 | name = str(tool_call.get("name") or "") |
| 92 | arguments = tool_call.get("args", {}) |
| 93 | |
| 94 | if isinstance(arguments, str): |
| 95 | try: |
| 96 | args = json.loads(arguments) |
| 97 | except json.JSONDecodeError: |
| 98 | args = {"arguments": arguments} |
| 99 | elif isinstance(arguments, dict): |
| 100 | args = arguments |
| 101 | else: |
| 102 | args = {} |
| 103 | |
| 104 | langchain_tool_calls.append( |
| 105 | { |
| 106 | "name": name, |
| 107 | "args": args, |
| 108 | "id": str(tool_call.get("id") or ""), |
| 109 | "type": "tool_call", |
| 110 | } |
| 111 | ) |
| 112 |
no outgoing calls