从原型到生产的鸿沟 LangChain让构建LLM Agent原型变得非常简单——几十行代码就能实现一个能调用工具、检索知识的智能体。但从原型到生产环境，需要解决可靠性、性能、可观测性、成本控制等一系列工程问题。
基础架构 Agent框架选择 from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_openai import ChatOpenAI from langchain.tools import Tool from langchain.memory import ConversationBufferWindowMemory # 使用工具调用Agent（比ReFi更可靠） llm = ChatOpenAI( model="qwen3-32b", base_url="http://localhost:8000/v1", temperature=0.1, # 生产环境低温度 max_retries=3, timeout=30 ) # 工具定义 tools = [ Tool( name="search", func=search_function, description="搜索知识库中的信息" ), Tool( name="calculator", func=calculator_function, description="数学计算" ), ] # 记忆管理 memory = ConversationBufferWindowMemory( memory_key="chat_history", k=10, # 只保留最近10轮对话 return_messages=True ) agent = create_tool_calling_agent(llm, tools, prompt) executor = AgentExecutor( agent=agent, tools=tools, memory=memory, max_iterations=5, # 限制迭代次数 max_execution_time=60, # 超时60秒 early_stopping_method="generate", verbose=True ) 可靠性工程 错误处理与重试 from tenacity import retry, stop_after_attempt, wait_exponential import logging logger = logging.getLogger(__name__) class RobustAgentExecutor: def __init__(self, agent_executor, fallback_response="抱歉，我暂时无法处理这个请求。"): self.executor = agent_executor self.fallback = fallback_response @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry_error_callback=lambda _: None ) async def invoke(self, input_data): try: result = await self.executor.ainvoke(input_data) # 验证结果 if not result or "output" not in result: raise ValueError("Invalid agent output") return result except TimeoutError: logger.warning(f"Agent timeout for input: {input_data}") return {"output": self.fallback} except Exception as e: logger.error(f"Agent error: {e}", exc_info=True) raise async def safe_invoke(self, input_data): """不抛异常的调用""" result = await self.invoke(input_data) return result or {"output": self.fallback} 工具调用验证 from pydantic import BaseModel, validator class SearchInput(BaseModel): query: str max_results: int = 5 @validator('query') def query_must_be_valid(cls, v): if not v or len(v.strip()) < 2: raise ValueError("Query too short") if len(v) > 500: raise ValueError("Query too long") return v.strip() @validator('max_results') def max_results_range(cls, v): if v < 1 or v > 20: raise ValueError("max_results must be 1-20") return v class ValidatedTool: def __init__(self, func, input_schema): self.func = func self.input_schema = input_schema async def __call__(self, **kwargs): # 验证输入 validated = self.input_schema(**kwargs) try: result = await self.func(**validated.dict()) # 验证输出 if not result: return "No results found" return result except Exception as e: logger.error(f"Tool error: {e}") return f"Tool execution failed: {str(e)}" 速率限制 import asyncio from datetime import datetime, timedelta class RateLimiter: def __init__(self, max_calls=10, window_seconds=60): self.max_calls = max_calls self.window = window_seconds self.calls = [] self.lock = asyncio.Lock() async def acquire(self): async with self.lock: now = datetime.now() # 清理过期记录 self.calls = [t for t in self.calls if now - t < timedelta(seconds=self.window)] if len(self.calls) >= self.max_calls: wait_time = self.window - (now - self.calls[0]).total_seconds() await asyncio.sleep(wait_time) self.calls.append(now) class RateLimitedAgent: def __init__(self, executor, rate_limiter): self.executor = executor self.limiter = rate_limiter async def invoke(self, input_data): await self.limiter.acquire() return await self.executor.ainvoke(input_data) 可观测性 链路追踪 from langchain.callbacks import BaseCallbackHandler import json import time class TracingCallbackHandler(BaseCallbackHandler): def __init__(self): self.traces = [] self.current_trace = None def on_chain_start(self, serialized, inputs, **kwargs): self.current_trace = { "chain": serialized.get("name", "unknown"), "start_time": time.time(), "inputs": str(inputs)[:500], "steps": [] } def on_llm_start(self, serialized, prompts, **kwargs): if self.current_trace: self.current_trace["steps"].append({ "type": "llm", "model": serialized.get("name", "unknown"), "start_time": time.time() }) def on_llm_end(self, response, **kwargs): if self.current_trace and self.current_trace["steps"]: step = self.current_trace["steps"][-1] step["end_time"] = time.time() step["duration"] = step["end_time"] - step["start_time"] step["tokens"] = response.llm_output.get("token_usage", {}) def on_tool_start(self, serialized, input_str, **kwargs): if self.current_trace: self.current_trace["steps"].append({ "type": "tool", "tool": serialized.get("name", "unknown"), "input": input_str[:200], "start_time": time.time() }) def on_tool_end(self, output, **kwargs): if self.current_trace and self.current_trace["steps"]: step = self.current_trace["steps"][-1] step["end_time"] = time.time() step["duration"] = step["end_time"] - step["start_time"] step["output"] = str(output)[:500] def on_chain_end(self, outputs, **kwargs): if self.current_trace: self.current_trace["end_time"] = time.time() self.current_trace["duration"] = ( self.current_trace["end_time"] - self.current_trace["start_time"] ) self.current_trace["output"] = str(outputs)[:500] self.traces.append(self.current_trace) self.current_trace = None # 使用 tracing = TracingCallbackHandler() result = executor.invoke( {"input": "What is the weather?"}, config={"callbacks": [tracing]} ) 结构化日志 import structlog logger = structlog.get_logger() class LoggingMiddleware: async def log_request(self, request_data, response_data, duration): logger.info( "agent_request", input_length=len(str(request_data)), output_length=len(str(response_data)), duration_ms=duration * 1000, agent_version="1.0.0", timestamp=datetime.now().isoformat() ) 成本控制 Token预算管理 class TokenBudget: def __init__(self, daily_budget=1000000): self.daily_budget = daily_budget self.used = 0 self.date = datetime.now().date() def consume(self, tokens): # 跨天重置 if datetime.now().date() != self.date: self.used = 0 self.date = datetime.now().date() self.used += tokens if self.used > self.daily_budget: raise BudgetExceededError( f"Daily budget exceeded: {self.used}/{self.daily_budget}" ) def remaining(self): return self.daily_budget - self.used class BudgetAwareAgent: def __init__(self, executor, budget): self.executor = executor self.budget = budget async def invoke(self, input_data): # 预估输入token estimated_input = len(str(input_data)) // 4 if self.budget.remaining() < estimated_input + 1000: return {"output": "今日配额已用尽，请明天再试。"} result = await self.executor.ainvoke(input_data) # 记录实际消耗 if "intermediate_steps" in result: total_tokens = sum( step.get("token_usage", {}).get("total_tokens", 0) for step in result["intermediate_steps"] ) self.budget.consume(total_tokens) return result 部署架构 FastAPI服务 from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI(title="LLM Agent API") class ChatRequest(BaseModel): message: str session_id: str = None max_tokens: int = 2048 class ChatResponse(BaseModel): response: str session_id: str latency_ms: float @app.post("/chat", response_model=ChatResponse) async def chat(request: ChatRequest): start_time = time.time() try: result = await agent.safe_invoke({ "input": request.message, "session_id": request.session_id }) latency = (time.time() - start_time) * 1000 return ChatResponse( response=result["output"], session_id=request.session_id, latency_ms=latency ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) 结语 LangChain Agent的生产化需要从可靠性、可观测性、成本控制三个维度系统性地构建基础设施。通过错误处理、速率限制、链路追踪和预算管理，可以将原型级别的Agent转变为可靠的生产服务。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
