从原型到生产的鸿沟

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论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。