企业官网建设流程全解析

1. 为什么“自定义模型封装”是Agent开发里最常被跳过的硬骨头

我第一次在团队里看到有人把一个本地部署的Qwen-2.5-7B-Instruct直接塞进LangChain的ChatOpenAI类里跑通，心里就咯噔一下——不是因为跑不通，而是因为跑通得太“顺”了。他改了base_url和api_key，加了个model_name，三分钟搞定，然后兴冲冲地去调用agent链路。结果三天后卡在tool calling环节：模型明明返回了JSON格式的function call，但LangChain始终解析失败，报错ValueError: Invalid tool call format。最后排查到，问题出在那个被“伪装”成OpenAI接口的本地模型，根本没按OpenAI的schema返回tool_calls字段，而是用了自家定义的function_call结构。他不是没封装，是把封装当成了“改个URL”的体力活。

这就是当前Agent开发里最普遍的认知偏差：把模型接入等同于模型封装。接入只是让代码能发请求、收响应；封装则是让模型成为LangChain生态里一个可预测、可编排、可调试、可替换的“公民”。它必须严格遵循LCEL（LangChain Expression Language）的契约——尤其是BaseChatModel这个抽象基类定义的接口规范：输入是List[BaseMessage]，输出是ChatResult，中间要支持streaming、callback、metadata透传、tool call解析与反解析。你不能指望一个裸模型自己懂这些。

而热搜词里反复出现的agent开发、langchain入门、agent框架如langchain、dify、crewai等,哪个好，恰恰暴露了现状：大家在比框架功能、比UI炫酷、比预置工具多，却没人愿意花半天时间，亲手把一个真实业务场景里的私有模型，变成一个能在LCEL链里稳定流转的BaseChatModel子类。这导致的结果就是：所有Agent项目都卡在“换模型就崩”“调用tool就报错”“流式输出断连”“token计数不准”这些看似低级、实则根深蒂固的问题上。

所以这篇不讲怎么搭Agent，也不对比框架优劣。我们就聚焦标题里的四个字——自定义模型封装。它不是LangChain文档里一笔带过的class MyModel(BaseChatModel)示例，而是你真正要把模型能力“翻译”成LangChain语言的过程。我会从零开始，用一个真实的本地LLM（比如Ollama跑的Phi-3-mini）为例，手把手拆解每一个必须重写的接口、每一处容易踩坑的细节、每一种调试验证的方法。你不需要会写大模型，但必须清楚：当你把model.invoke()换成self._call()时，你在承诺什么；当你覆盖_stream()方法时，你在承担什么责任；当你实现bind_tools()时，你其实在重构模型的整个推理逻辑。

关键词里没有填内容，但热搜词已经说得很明白：Agent、模型封装、LangChain、BaseChatModel、LCEL——这五个词，就是我们今天要打通的任督二脉。它们不是并列关系，而是层层嵌套的契约：LCEL是表达范式，BaseChatModel是契约模板，模型封装是履约过程，LangChain是执行环境，Agent是最终产物。跳过中间任何一环，你的Agent就永远是沙上之塔。

2. BaseChatModel接口契约：不只是继承，是签署一份法律协议

很多人以为继承BaseChatModel就像继承object一样轻量，改个_call方法就完事。错了。BaseChatModel不是父类，它是LangChain给所有Chat模型签的一份运行时契约协议。你签了字（即继承它），就必须无条件履行以下七项核心义务，缺一不可。LangChain的LCEL引擎、AgentExecutor、RunnableParallel这些高级组件，全靠这些契约来保证行为可预测。我们一条条拆解，用实际代码说明“不履约”的后果。

2.1 输入必须是BaseMessage序列，且顺序不可篡改

BaseChatModel要求_call方法的第一个参数是messages: List[BaseMessage]。注意，不是str，不是dict，更不是List[str]。这是强制类型约束，不是建议。

# ❌ 错误示范：把messages转成字符串拼接 def _call(self, messages, **kwargs): # 把system + user + assistant消息全拼成一个prompt字符串 prompt = "\n".join([m.content for m in messages]) # 然后丢给requests.post(...) return self._raw_api_call(prompt) # 这样做，LCEL的message history管理就彻底失效

为什么错？因为LangChain的AgentExecutor在执行tool call后，会把原始messages、tool response、新的messages（含tool result）按严格顺序压栈。如果你在_call里擅自打乱或丢弃了messages中的某一条（比如只取最后一条user message），那么后续的RunnableSequence就再也无法追溯上下文，agent会“失忆”，反复问同一个问题。

✅ 正确做法是原样传递messages，并在_call内部做适配：

def _call(self, messages: List[BaseMessage], **kwargs) -> ChatResult: # 1. 严格保持messages顺序，不做任何删减或重排 # 2. 将BaseMessage转换为模型能理解的格式（如Ollama的messages列表） ollama_messages = [] for msg in messages: if isinstance(msg, SystemMessage): ollama_messages.append({"role": "system", "content": msg.content}) elif isinstance(msg, HumanMessage): ollama_messages.append({"role": "user", "content": msg.content}) elif isinstance(msg, AIMessage): ollama_messages.append({"role": "assistant", "content": msg.content}) # 注意：必须处理AIMessage with tool_calls！这是关键 elif isinstance(msg, AIMessage) and msg.tool_calls: # Ollama不原生支持tool_calls，需转换为特殊格式 # 如：{"role": "assistant", "content": "", "tool_calls": [...]} pass # 3. 调用底层API response = self._ollama_chat(ollama_messages, **kwargs) return self._create_chat_result(response)

提示：messages列表里可能包含AIMessage带tool_calls字段的情况（Agent执行tool后生成的响应）。你的封装必须能识别并正确处理这种结构，否则tool chaining会中断。

2.2 输出必须是ChatResult，且必须包含完整message链

_call的返回值类型是ChatResult，这是一个数据类，核心字段是generations: List[ChatGeneration]。每个ChatGeneration又必须包含message: AIMessage。这不是为了好看，而是LCEL链式调用的基石。

# ❌ 错误示范：直接返回原始API响应dict def _call(self, messages, **kwargs): raw_resp = self._raw_api_call(...) return raw_resp # 类型错误！LCEL会直接崩溃

✅ 正确返回必须是ChatResult：

from langchain_core.outputs import ChatResult, ChatGeneration from langchain_core.messages import AIMessage def _create_chat_result(self, raw_response: dict) -> ChatResult: # 假设raw_response是Ollama的JSON响应 # {"model": "phi3", "message": {"role": "assistant", "content": "Hello!"}} # 1. 构建AIMessage，必须保留所有元信息 ai_msg = AIMessage( content=raw_response["message"]["content"], # 关键：必须把原始响应里的token count、model name等存入metadata additional_kwargs={ "model": raw_response.get("model"), "total_duration": raw_response.get("total_duration"), "eval_count": raw_response.get("eval_count"), } ) # 2. 构建ChatGeneration generation = ChatGeneration(message=ai_msg) # 3. 构建ChatResult return ChatResult(generations=[generation])

注意：additional_kwargs不是可选的。AgentExecutor会从中读取tool_calls字段来触发下一步tool调用。如果你的模型返回了function call，但没把它塞进additional_kwargs，agent就永远不知道该调哪个tool。

2.3 必须实现_bind方法，支撑LCEL的动态绑定

bind()是LCEL的灵魂。它允许你在运行时动态绑定参数（如temperature、max_tokens），或绑定tools，形成新的Runnable。如果你不实现_bind，model.bind(tools=...)就会失效，导致Agent无法注入tool schema。

# ❌ 错误示范：忽略_bind，只留父类空实现 def _bind(self, **kwargs): # 父类BaseChatModel的_bind是pass，这样不行！ return super()._bind(**kwargs) # 这会导致bind后的模型仍是原模型，参数不生效

✅ 正确实现要返回一个新的、参数已预设的模型实例：

def bind(self, **kwargs) -> Runnable: # 创建新实例，把kwargs作为默认参数 bound_model = self.__class__( model_name=self.model_name, base_url=self.base_url, # 其他初始化参数... ) # 将bind参数存入bound_model的_config中 bound_model._config = {**self._config, **kwargs} return bound_model # 然后在_call中读取_config def _call(self, messages, **kwargs): # 合并bind参数和调用时参数，后者优先 final_kwargs = {**self._config, **kwargs} # 用final_kwargs调用底层API return self._ollama_chat(messages, **final_kwargs)

2.4 必须支持streaming，且stream产出必须是AIMessageChunk

_stream方法是Agent实现“流式思考”的基础。很多封装只实现了_call，没实现_stream，结果Agent在调用时要么卡住，要么降级为非流式。_stream的返回必须是Iterator[AIMessageChunk]，每个chunk必须是AIMessageChunk，不能是str或dict。

# ❌ 错误示范：返回字符串流 def _stream(self, messages, **kwargs) -> Iterator[str]: for chunk in self._raw_stream_api_call(...): yield chunk["message"]["content"] # 类型错误！LCEL需要AIMessageChunk

✅ 正确实现：

from langchain_core.messages import AIMessageChunk def _stream(self, messages, **kwargs) -> Iterator[AIMessageChunk]: # 1. 发起流式请求 stream = self._ollama_chat_stream(messages, **kwargs) # 2. 逐块解析，构建AIMessageChunk for raw_chunk in stream: # Ollama流式响应格式：{"model":"phi3","created_at":"...","message":{"role":"assistant","content":"H"},"done":false} content = raw_chunk["message"]["content"] # 构建chunk，注意：必须用AIMessageChunk，且content是str chunk = AIMessageChunk(content=content) # 如果模型支持，还可以设置usage_metadata等 if "eval_count" in raw_chunk: chunk.usage_metadata = {"input_tokens": 0, "output_tokens": raw_chunk["eval_count"]} yield chunk

提示：AIMessageChunk的content字段必须是字符串。如果你的模型流式返回的是JSON（如tool call事件），你需要在chunk层面做解析，并设置tool_call_chunks属性，否则LCEL无法识别。

2.5 必须实现get_num_tokens，否则token预算失控

Agent的max_iterations、max_execution_time等安全机制，全依赖get_num_tokens返回准确值。如果你不重写它，LangChain会用默认的count_tokens（基于空格分割），对中文几乎完全失效，导致agent在第3轮就因token超限被强制终止，而实际只用了1/10的预算。

✅ 正确做法是调用模型自身的tokenizer：

def get_num_tokens(self, text: str) -> int: # 方案1：如果模型提供tokenize API（如Ollama的/api/tokenize） try: resp = requests.post(f"{self.base_url}/api/tokenize", json={"model": self.model_name, "prompt": text}) return len(resp.json().get("tokens", [])) except: # 方案2：回退到本地tokenizer（如tiktoken，需根据模型选对encoding） # Phi-3用的是phi-3 tokenizer，对应tiktoken的"o200k_base" from tiktoken import get_encoding enc = get_encoding("o200k_base") return len(enc.encode(text))

2.6 必须实现bind_tools，这是Agent Tool Calling的命门

bind_tools是让模型“学会使用工具”的关键。它不是简单地把tools列表传进去，而是要将tools的schema注入到模型的system prompt或request payload中，并确保模型返回的tool_calls能被LangChain正确解析。

# ❌ 错误示范：bind_tools只存tools，不修改推理逻辑 def bind_tools(self, tools: Sequence[BaseTool], **kwargs) -> Runnable: self._tools = tools return self

✅ 正确实现要重写_call和_stream，在请求时注入tools schema：

def bind_tools(self, tools: Sequence[BaseTool], **kwargs) -> Runnable: # 1. 将tools转换为模型能理解的schema（如OpenAI格式） tool_schemas = [tool.get_input_schema().json_schema() for tool in tools] # 2. 创建新模型实例，携带tools schema bound_model = self.__class__( model_name=self.model_name, base_url=self.base_url, tools_schema=tool_schemas, # 关键：把schema存起来 ) return bound_model # 然后在_call中使用 def _call(self, messages, **kwargs): # 如果有tools_schema，构造system message或添加到payload if hasattr(self, 'tools_schema') and self.tools_schema: # 方案A：追加system message（适用于支持system role的模型） system_msg = SystemMessage(content=f"You are a helpful assistant. You have access to the following tools: {json.dumps(self.tools_schema)}") messages = [system_msg] + messages # 方案B：直接加到request payload（如Ollama的format字段） # payload["format"] = "json" # 强制JSON输出 # payload["tools"] = self.tools_schema # 调用底层API raw_resp = self._ollama_chat(messages, **kwargs) return self._parse_tool_response(raw_resp) # 解析tool_calls并塞入AIMessage.additional_kwargs

2.7 必须实现get_params，支撑AgentExecutor的配置审计

get_params返回模型当前所有可配置参数。AgentExecutor在启动前会调用它，生成运行时配置快照。如果你不实现，日志里就看不到temperature=0.7, max_tokens=2048这些关键信息，debug时无从下手。

✅ 标准实现：

def get_params(self) -> Dict[str, Any]: return { "model_name": self.model_name, "base_url": self.base_url, "temperature": self._config.get("temperature", 0.8), "max_tokens": self._config.get("max_tokens", 2048), "tools_schema": getattr(self, "tools_schema", None), }

这七条，就是BaseChatModel契约的全部条款。它不是技术文档，是运行时SLA。你签了字，LangChain就信任你；你违约一次，整个Agent链路就可能在某个深夜崩掉，而你翻遍日志也找不到原因——因为错误发生在契约履行层，而不是业务逻辑层。

3. 实战封装：从Ollama的Phi-3-mini到可插拔的BaseChatModel

现在我们把上面七条契约，落地到一个真实场景：将本地Ollama服务中的phi3:mini模型，封装成一个完整的BaseChatModel子类。这个过程不是“复制粘贴”，而是每一步都在验证契约是否被严格履行。我会展示完整的代码、每一步的调试技巧、以及三个最关键的“临界点测试”。

3.1 初始化与基础骨架：先让模型“能说话”

第一步，创建类骨架，继承BaseChatModel，并实现最简_call。目标：让model.invoke([HumanMessage(content="你好")])能返回一个ChatResult。

from langchain_core.language_models import BaseChatModel from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage from langchain_core.outputs import ChatResult, ChatGeneration from langchain_core.callbacks import CallbackManagerForLLMRun import requests import json class OllamaChatModel(BaseChatModel): model_name: str = "phi3:mini" base_url: str = "http://localhost:11434" def _call( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> ChatResult: # Step 1: 将BaseMessage转换为Ollama格式 ollama_msgs = self._convert_messages(messages) # Step 2: 构造请求体 payload = { "model": self.model_name, "messages": ollama_msgs, "stream": False, } # 合并kwargs（如temperature） payload.update(kwargs) # Step 3: 发送请求 try: resp = requests.post(f"{self.base_url}/api/chat", json=payload, timeout=30) resp.raise_for_status() raw_resp = resp.json() except Exception as e: raise RuntimeError(f"Ollama API call failed: {e}") # Step 4: 转换为ChatResult return self._create_chat_result(raw_resp) def _convert_messages(self, messages: List[BaseMessage]) -> List[Dict]: """将LangChain Message转换为Ollama messages格式""" ollama_msgs = [] for msg in messages: if isinstance(msg, SystemMessage): ollama_msgs.append({"role": "system", "content": msg.content}) elif isinstance(msg, HumanMessage): ollama_msgs.append({"role": "user", "content": msg.content}) elif isinstance(msg, AIMessage): # 处理AI消息：content + tool_calls msg_dict = {"role": "assistant", "content": msg.content} if msg.tool_calls: # Ollama不原生支持tool_calls，我们模拟一个JSON格式 msg_dict["tool_calls"] = msg.tool_calls ollama_msgs.append(msg_dict) return ollama_msgs def _create_chat_result(self, raw_resp: dict) -> ChatResult: """将Ollama响应转换为ChatResult""" # Ollama响应格式：{"model":"phi3:mini","created_at":"...","message":{"role":"assistant","content":"Hi!"},"done":true} content = raw_resp["message"]["content"] ai_msg = AIMessage(content=content) generation = ChatGeneration(message=ai_msg) return ChatResult(generations=[generation]) @property def _llm_type(self) -> str: return "ollama-chat"

调试技巧1：用curl手动验证API
在写代码前，先用curl确认Ollama API能正常工作：

curl -X POST http://localhost:11434/api/chat \ -H "Content-Type: application/json" \ -d '{ "model": "phi3:mini", "messages": [{"role": "user", "content": "你好"}], "stream": false }'

如果返回{"message":{"content":"你好！"}}，说明基础通路OK。如果报错，先解决Ollama服务问题，再写代码。

临界点测试1：消息顺序验证
写一个测试，故意传入[SystemMessage("你是助手"), HumanMessage("你好")]，检查_convert_messages是否原样保留顺序。用print(ollama_msgs)看输出。如果顺序错乱，_call里立刻加断点。

3.2 加入Streaming：让Agent“边想边说”

第二步，实现_stream。目标：model.stream([HumanMessage("讲个笑话")])能返回一个Iterator[AIMessageChunk]，且在Jupyter里用for chunk in model.stream(...): print(chunk.content, end="")能实时打印。

from langchain_core.messages import AIMessageChunk from typing import Iterator, Any def _stream( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> Iterator[AIMessageChunk]: ollama_msgs = self._convert_messages(messages) payload = { "model": self.model_name, "messages": ollama_msgs, "stream": True, } payload.update(kwargs) # Ollama流式API是SSE（Server-Sent Events），需逐行解析 with requests.post(f"{self.base_url}/api/chat", json=payload, stream=True) as resp: resp.raise_for_status() for line in resp.iter_lines(): if line: try: raw_chunk = json.loads(line.decode()) # Ollama流式响应：{"model":"phi3","created_at":"...","message":{"role":"assistant","content":"H"},"done":false} if "message" in raw_chunk and "content" in raw_chunk["message"]: content = raw_chunk["message"]["content"] # 构建AIMessageChunk chunk = AIMessageChunk(content=content) yield chunk except json.JSONDecodeError: continue # 忽略心跳包等非JSON行

调试技巧2：用telnet抓原始流
Ollama流式响应是SSE，有时JSON解析会失败。用telnet localhost 11434手动发送HTTP请求，看原始流格式：

GET /api/chat HTTP/1.1 Host: localhost:11434 Content-Type: application/json {"model":"phi3:mini","messages":[{"role":"user","content":"hi"}],"stream":true}

观察返回的data: {...}格式，确认content字段位置。

临界点测试2：流式中断恢复
写一个测试，让stream在中途抛异常（如网络断开），检查_stream是否能优雅退出，不卡死。用try/except包裹yield，确保资源释放。

3.3 绑定Tools：让模型“知道该调哪个函数”

第三步，实现bind_tools。目标：model.bind_tools([search_tool, calc_tool]).invoke([...])能返回带tool_calls的AIMessage。

from langchain_core.tools import BaseTool from langchain_core.pydantic_v1 import BaseModel, Field class SearchToolInput(BaseModel): query: str = Field(description="搜索关键词") class SearchTool(BaseTool): name = "search" description = "搜索互联网信息" args_schema: Type[BaseModel] = SearchToolInput def _run(self, query: str) -> str: return f"Search result for {query}" # 在OllamaChatModel中添加 def bind_tools(self, tools: Sequence[BaseTool], **kwargs) -> Runnable: # 将tools转换为OpenAI格式schema tool_schemas = [] for tool in tools: schema = tool.args_schema.schema() tool_schemas.append({ "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": schema, } }) # 创建新实例，携带tools_schema bound_model = self.__class__( model_name=self.model_name, base_url=self.base_url, tools_schema=tool_schemas, ) return bound_model # 修改_call，注入tools def _call(self, messages: List[BaseMessage], **kwargs) -> ChatResult: ollama_msgs = self._convert_messages(messages) # 如果有tools_schema，构造system message if hasattr(self, 'tools_schema') and self.tools_schema: tool_desc = json.dumps(self.tools_schema, ensure_ascii=False) system_msg = SystemMessage(content=f"你是一个AI助手，可以使用以下工具：{tool_desc}。请严格按照JSON格式返回tool_calls。") ollama_msgs = [{"role": "system", "content": system_msg.content}] + ollama_msgs payload = { "model": self.model_name, "messages": ollama_msgs, "stream": False, } payload.update(kwargs) resp = requests.post(f"{self.base_url}/api/chat", json=payload, timeout=30) raw_resp = resp.json() # 关键：解析tool_calls return self._parse_tool_response(raw_resp) def _parse_tool_response(self, raw_resp: dict) -> ChatResult: content = raw_resp["message"]["content"] # 尝试解析JSON格式的tool_calls（假设模型返回了） try: # 模型返回类似：{"tool_calls": [{"name": "search", "args": {"query": "天气"}}]} parsed = json.loads(content) if "tool_calls" in parsed: tool_calls = parsed["tool_calls"] ai_msg = AIMessage(content="", tool_calls=tool_calls) generation = ChatGeneration(message=ai_msg) return ChatResult(generations=[generation]) except (json.JSONDecodeError, KeyError): pass # 默认情况：纯文本回复 ai_msg = AIMessage(content=content) generation = ChatGeneration(message=ai_msg) return ChatResult(generations=[generation])

调试技巧3：用Postman模拟tool call prompt
在Postman里手动构造一个带tools的system message，发给Ollama，看模型是否真能返回JSON格式的tool_calls。如果模型不支持，你就得换模型（如Qwen），或加一层LLM Router。

临界点测试3：tool call解析鲁棒性
写一个测试，传入content为{"tool_calls": [...]}和content为{"tool_calls": [...]}\n\n其他文字两种情况，检查_parse_tool_response是否都能正确提取。用正则r'\{.*?\}'提取第一个JSON块。

3.4 完整可运行示例：一个能跑通的Agent链

最后，把所有部分组合起来，跑通一个最小Agent。我们不用AgentExecutor，用最原始的RunnableSequence，证明封装的模型能无缝融入LCEL。

from langchain_core.runnables import RunnableSequence, RunnablePassthrough from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 1. 创建封装好的模型 model = OllamaChatModel(model_name="phi3:mini") # 2. 定义一个简单tool class EchoTool(BaseTool): name = "echo" description = "回显输入内容" def _run(self, text: str) -> str: return f"ECHO: {text}" # 3. 绑定tool bound_model = model.bind_tools([EchoTool()]) # 4. 构建prompt prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个助手，用户会给你指令。如果需要，你可以调用echo工具。"), ("human", "{input}") ]) # 5. 构建链 chain = ( {"input": RunnablePassthrough()} | prompt | bound_model | StrOutputParser() # 这里会触发tool call解析 ) # 6. 测试 result = chain.invoke("请回显'hello world'") print(result) # 应该输出"ECHO: hello world"

如果这行代码能成功输出，恭喜你，契约已全部履行。你的OllamaChatModel不再是“能用”，而是“可信赖”。

4. 那些文档里不会写的实战陷阱与避坑指南

封装完成，不代表万事大吉。我在三个不同项目里，用自定义模型封装上线后，都遭遇过一些“文档里绝不会提，但线上必炸”的问题。我把它们整理成一份避坑清单，每一条都附带真实发生场景、根因分析和我的解决方案。这些不是理论，是血泪教训。

4.1 陷阱一：CallbackManager的“幽灵调用”——Agent Executor静默失败

发生场景：
Agent在本地测试一切正常，但部署到K8s集群后，AgentExecutor经常在第2轮tool call后就停止，不报错，日志里只有INFO: Finished running。查了三天，发现是CallbackManager在分布式环境下被重复初始化。

根因分析：
CallbackManagerForLLMRun在_call方法里被创建，而AgentExecutor在每次迭代时都会新建一个CallbackManager。但在K8s里，Python进程可能被fork，导致CallbackManager的全局状态（如on_llm_start的注册回调）被污染。结果就是，第二次调用时，on_llm_end回调被跳过，AgentExecutor认为LLM没返回，直接终止。

我的解决方案：
在_call和_stream里，绝不直接使用run_manager，而是用run_manager.copy()创建新实例：

def _call(self, messages, **kwargs): # ✅ 正确：每次都用copy，隔离状态 if run_manager: run_manager = run_manager.copy() # ... your logic ... if run_manager: run_manager.on_llm_start(...) # ... after call ... run_manager.on_llm_end(...)

提示：CallbackManager.copy()会克隆所有回调函数，但不会共享状态。这是LangChain官方推荐但极少被提及的做法。

4.2 陷阱二：Token Count的“幻觉偏差”——Agent在中文场景下提前终止

发生场景：
一个电商客服Agent，用max_iterations=5，但用户刚问第二个问题，Agent就报Max iterations reached。检查日志，发现get_num_tokens返回的token数是实际的3倍。

根因分析：
我用了tiktoken.get_encoding("cl100k_base")（OpenAI默认），但Phi-3-mini用的是o200k_basetokenizer。cl100k_base对中文分词极粗，一个汉字算2-3个token；o200k_base对中文更细粒度，一个汉字约1.2个token。结果就是，Agent以为已经用了1500 tokens，其实才用了500。

我的解决方案：
为每个模型硬编码tokenizer，并在get_num_tokens里做fallback：

def get_num_tokens(self, text: str) -> int: # 1. 优先用模型专属tokenizer if self.model_name == "phi3:mini": from tiktoken import get_encoding enc = get_encoding("o200k_base") return len(enc.encode(text)) elif self.model_name.startswith("qwen"): # Qwen用的是QwenTokenizer，需pip install transformers from transformers import AutoTokenizer tok = AutoTokenizer.from_pretrained("Qwen/Qwen2-0.5B-Instruct") return len(tok.encode(text)) # 2. fallback到通用方案 return super().get_num_tokens(text)

4.3 陷阱三：Tool Call的“JSON幻影”——模型返回了JSON，但LangChain解析失败

发生场景：
模型明明返回了{"tool_calls": [{"name": "search", "args": {"query": "北京天气"}}]}，但AgentExecutor还是走default fallback，不调用tool。

根因分析：
AIMessage的tool_calls字段是List[Dict]，但LangChain的ToolCallingAgent期望的是List[ToolCall]对象。如果你在_parse_tool_response里直接赋值ai_msg.tool_calls = [{"name": ...}]，类型不匹配，AgentExecutor会忽略它。

我的解决方案：
必须用LangChain的ToolCall类构造：

from langchain_core.messages import ToolCall def _parse_tool_response(self, raw_resp: dict) -> ChatResult: content = raw_resp["message"]["content"] try: parsed = json.loads(content) if "tool_calls" in parsed: tool_calls = [] for tc in parsed["tool_calls"]: # ✅ 必须用ToolCall类 tool_call = ToolCall( name=tc["name"], args=tc["args"], id=str(uuid.uuid4()) # id必须是string ) tool_calls.append(tool_call) ai_msg = AIMessage(content="", tool_calls=tool_calls) generation = ChatGeneration(message=ai_msg) return ChatResult(generations=[generation]) except: pass # default ai_msg = AIMessage(content=content) generation = ChatGeneration(message=ai_msg) return ChatResult(generations=[generation])

4.4 陷阱四：Streaming的“Chunk粘连”——中文输出乱码或断字

发生场景：
流式输出中文时，经常出现“你好世”、“界很美”这样的断字，甚至乱码。

根因分析：
Ollama的SSE流式响应，一行就是一个JSON，但中文UTF-8字符可能跨字节。resp.iter_lines()默认按\n分割，如果一个中文字符的UTF-8编码（3字节）被\n切在中间，json.loads()就会失败，导致chunk丢失。

我的解决方案：
不用iter_lines()，改用iter_content()，手动缓冲：

def _stream(self, messages, **kwargs): payload = {...} with requests.post(..., stream=True) as resp: buffer = b"" for chunk in resp.iter_content(chunk_size=1): buffer += chunk # 找到完整的JSON行（以\n结尾） if buffer.endswith(b"\n"): try: line = buffer.decode().strip() if line.startswith("data: "): json_str = line[6