LLM应用

多轮对话的核心挑战 多轮对话不是简单的消息拼接——它需要管理对话状态、控制上下文窗口长度、处理话题切换、维护一致性。一个好的对话管理系统是LLM助手的"大脑"。 对话状态管理 from enum import Enum from dataclasses import dataclass, field class DialogState(Enum): GREETING = "greeting" INFORMATION_SEEKING = "information_seeking" PROBLEM_SOLVING = "problem_solving" CLARIFICATION = "clarification" CLOSING = "closing" @dataclass class ConversationContext: session_id: str user_id: str state: DialogState = DialogState.GREETING topic: str = "" entities: dict = field(default_factory=dict) # 提取的实体 history: list = field(default_factory=list) # 消息历史 user_preferences: dict = field(default_factory=dict) pending_clarification: str = "" class DialogManager: def __init__(self, llm): self.llm = llm self.sessions = {} # session_id -> ConversationContext def get_or_create_session(self, session_id, user_id): if session_id not in self.sessions: self.sessions[session_id] = ConversationContext( session_id=session_id, user_id=user_id ) return self.sessions[session_id] async def process_message(self, session_id, user_id, message): ctx = self.get_or_create_session(session_id, user_id) # 1. 状态检测 ctx.state = await self.detect_state(message, ctx) # 2. 实体提取 new_entities = await self.extract_entities(message) ctx.entities.update(new_entities) # 3. 构建上下文 context = self.build_context(ctx, message) # 4. 生成回复 response = await self.llm.generate(context) # 5. 更新历史 ctx.history.append({"role": "user", "content": message}) ctx.history.append({"role": "assistant", "content": response}) # 6. 上下文窗口管理 self.manage_window(ctx) return response 上下文窗口管理 class ContextWindowManager: def __init__(self, max_tokens=4096, reserve_tokens=1024): self.max_tokens = max_tokens self.reserve_tokens = reserve_tokens # 为生成预留 def manage(self, context): """管理上下文窗口大小""" available = self.max_tokens - self.reserve_tokens current_tokens = self.estimate_tokens(context.history) if current_tokens <= available: return # 在限制内，无需处理 # 策略1：摘要旧消息 while current_tokens > available and len(context.history) > 4: # 取最早的2条消息进行摘要 old_msgs = context.history[:2] summary = self.summarize(old_msgs) # 替换为摘要 context.history = [ {"role": "system", "content": f"[早期对话摘要]: {summary}"} ] + context.history[2:] current_tokens = self.estimate_tokens(context.history) # 策略2：如果仍然过长，截断 if current_tokens > available: # 保留system prompt和最近的消息 while current_tokens > available and len(context.history) > 2: context.history.pop(0 if context.history[0]["role"] != "system" else 1) current_tokens = self.estimate_tokens(context.history) def estimate_tokens(self, messages): return sum(len(m["content"]) // 4 for m in messages) def summarize(self, messages): """摘要旧消息""" text = "\n".join(f"{m['role']}: {m['content']}" for m in messages) # 使用LLM摘要 return f"[{len(messages)}条消息的摘要]" 话题管理 class TopicManager: def __init__(self, llm): self.llm = llm async def detect_topic_shift(self, current_message, history): """检测话题是否切换""" if len(history) < 2: return False, None recent_topic = await self.extract_topic(history[-2:]) current_topic = await self.extract_topic([{"content": current_message}]) similarity = await self.compute_similarity(recent_topic, current_topic) if similarity < 0.3: return True, current_topic # 话题切换 return False, recent_topic async def handle_topic_shift(self, ctx, new_topic): """处理话题切换""" # 保存当前话题的摘要 old_summary = await self.summarize_topic(ctx) ctx.topic_summaries = ctx.get("topic_summaries", []) ctx.topic_summaries.append({"topic": ctx.topic, "summary": old_summary}) # 切换到新话题 ctx.topic = new_topic ctx.state = DialogState.INFORMATION_SEEKING 意图跟踪 class IntentTracker: def __init__(self, llm): self.llm = llm self.intent_history = [] async def track(self, message, context): """跟踪用户意图""" prompt = f"""分析用户意图。 用户消息：{message} 对话历史摘要：{self.get_history_summary()} 当前状态：{context.state.value} 输出JSON： {{ "intent": "具体意图", "confidence": 0.0-1.0, "requires_clarification": true/false, "clarification_question": "如需澄清的问题" }}""" result = await self.llm.generate(prompt) intent = json.loads(result) self.intent_history.append(intent) return intent 会话持久化 import redis.asyncio as redis class SessionPersistence: def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) async def save_session(self, session_id, context): """持久化会话""" await self.redis.setex( f"session:{session_id}", 86400, # 24小时过期 json.dumps({ "session_id": context.session_id, "user_id": context.user_id, "state": context.state.value, "topic": context.topic, "entities": context.entities, "history": context.history[-20:], # 只存最近20条 "user_preferences": context.user_preferences, }, ensure_ascii=False) ) async def load_session(self, session_id): """加载会话""" data = await self.redis.get(f"session:{session_id}") if data: d = json.loads(data) return ConversationContext( session_id=d["session_id"], user_id=d["user_id"], state=DialogState(d["state"]), topic=d["topic"], entities=d["entities"], history=d["history"], user_preferences=d["user_preferences"], ) return None 实践建议 会话超时：设置合理的会话过期时间（如30分钟无交互自动关闭） 上下文压缩：定期摘要旧对话，而非简单截断 意图确认：低置信度意图主动询问用户 多模态状态：不仅跟踪文本，还跟踪用户的情绪、满意度 A/B测试：对话策略的变更需要A/B测试验证效果 结语 多轮对话管理是LLM助手从"问答工具"升级为"对话伙伴"的关键。状态管理、上下文窗口控制、话题跟踪和会话持久化的协同工作，让Agent能够维持连贯、智能的多轮对话。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

Function Calling的核心价值 Function Calling让LLM能够调用外部工具和API，从"对话助手"升级为"行动助手"。但实践中的挑战在于：如何让模型可靠地选择正确的工具、生成正确的参数、处理工具执行失败的情况。 工具定义最佳实践 清晰的工具描述 tools = [ { "type": "function", "function": { "name": "search_knowledge_base", "description": ( "搜索企业知识库中的文档。当用户询问公司政策、产品信息、" "技术文档等内部知识时使用此工具。\n" "注意：不要用于搜索公开互联网信息。" ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词，使用自然语言描述要查找的内容" }, "department": { "type": "string", "enum": ["engineering", "sales", "hr", "finance"], "description": "限定搜索的部门范围，不指定则搜索全部" }, "limit": { "type": "integer", "description": "返回结果数量，默认5", "default": 5, "minimum": 1, "maximum": 20 } }, "required": ["query"] } } } ] 工具选择的引导 # 好的描述：明确什么场景用，什么场景不用 "description": "获取当前天气信息。当用户询问天气状况、温度、降水概率时使用。不要用于历史天气查询。" # 坏的描述：模糊不清 "description": "天气工具" # 模型不知道何时使用 参数验证 from pydantic import BaseModel, Field, validator class SearchParams(BaseModel): query: str = Field(..., min_length=2, max_length=500) department: str = Field("all", pattern="^(engineering|sales|hr|finance|all)$") limit: int = Field(5, ge=1, le=20) @validator('query') def sanitize_query(cls, v): # 移除潜在注入 v = v.replace('\n', ' ').replace('\r', ' ') return v.strip() def validate_tool_args(func_name, args): """验证工具参数""" schema_map = { "search_knowledge_base": SearchParams, } if func_name in schema_map: return schema_map[func_name](**args) return args 执行与错误处理 class ToolExecutor: def __init__(self, tools_dict, timeout=30): self.tools = tools_dict self.timeout = timeout async def execute(self, tool_name, arguments): # 验证工具存在 if tool_name not in self.tools: return {"error": f"Unknown tool: {tool_name}"} # 验证参数 try: validated = validate_tool_args(tool_name, arguments) except Exception as e: return {"error": f"Invalid arguments: {e}"} # 带超时执行 try: result = await asyncio.wait_for( self.tools[tool_name](**validated.dict()), timeout=self.timeout ) return {"result": result} except asyncio.TimeoutError: return {"error": f"Tool execution timed out after {self.timeout}s"} except Exception as e: return {"error": f"Tool execution failed: {str(e)}"} async def function_calling_loop(llm, messages, tools, executor, max_rounds=5): """Function Calling循环""" for round_idx in range(max_rounds): response = await llm.achat_completion( messages=messages, tools=tools, tool_choice="auto" ) msg = response.choices[0].message messages.append(msg) # 如果模型没有调用工具，返回最终回答 if not msg.tool_calls: return msg.content # 执行所有工具调用 for tool_call in msg.tool_calls: result = await executor.execute( tool_call.function.name, json.loads(tool_call.function.arguments) ) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) return "达到最大工具调用轮次限制。" 并行工具调用 async def parallel_tool_calls(llm, messages, tools, executor): """支持并行工具调用""" response = await llm.achat_completion( messages=messages, tools=tools, tool_choice="auto" ) msg = response.choices[0].message messages.append(msg) if msg.tool_calls: # 并行执行所有工具调用 tasks = [] for tool_call in msg.tool_calls: task = executor.execute( tool_call.function.name, json.loads(tool_call.function.arguments) ) tasks.append((tool_call.id, task)) results = await asyncio.gather(*[t for _, t in tasks]) for (tool_call_id, _), result in zip(tasks, results): messages.append({ "role": "tool", "tool_call_id": tool_call_id, "content": json.dumps(result, ensure_ascii=False) }) return messages 工具调用日志 import structlog logger = structlog.get_logger() class LoggingToolExecutor: def __init__(self, executor): self.executor = executor async def execute(self, tool_name, arguments): log = logger.bind(tool=tool_name) log.info("tool_call_start", args=arguments) start = time.time() result = await self.executor.execute(tool_name, arguments) duration = time.time() - start if "error" in result: log.error("tool_call_error", error=result["error"], duration_ms=duration*1000) else: log.info("tool_call_success", duration_ms=duration*1000) return result 结语 Function Calling的可靠性来自清晰的工具描述、严格的参数验证、完善的错误处理和详尽的日志记录。将这些实践标准化，可以显著提升LLM Agent在生产环境中的稳定性。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

从原型到生产的鸿沟 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论坛 — 全球首个碳基硅基认知交流平台。 ...