生产级大模型Agent系统的工程化构建实践

from typing import TypedDict, List, Tuple, Any, Optional
from typing import Literal

class AgentState(TypedDict, total=False):
    # 用户当前输入
    user_input: str
    # 对话历史（符合OpenAI消息格式）
    chat_history: List[dict]
    # 执行轨迹：[("action_type", "action_name", result), ...]
    intermediate_steps: List[Tuple[str, str, Any]]
    # 最终输出
    final_output: Optional[str]
    # 重试计数
    retry_count: int
    # 当前步骤编号
    step_number: int

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
from langchain_openai import ChatOpenAI

# 初始化LLM和工具集
llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
tools = [web_search_tool, code_interpreter_tool, database_query_tool]
tool_executor = ToolExecutor(tools)

def build_messages(state: AgentState) -> List[dict]:
    """将状态转换为LLM消息列表"""
    messages = [{"role": "system", "content": system_prompt}]
    messages.extend(state.get("chat_history", []))
    return messages

def agent_node(state: AgentState) -> dict:
    """LLM推理节点：决定下一步行动或直接回答"""
    messages = build_messages(state)
    response = llm.bind_tools(tools).invoke(messages)
    
    if hasattr(response, "tool_calls") and response.tool_calls:
        return {
            "intermediate_steps": [("llm_decision", response.tool_calls[0]["name"], response.tool_calls[0])]
        }
    return {"final_output": response.content}

def tool_node(state: AgentState) -> dict:
    """工具执行节点：调用具体工具并返回结果"""
    last_step = state["intermediate_steps"][-1]
    tool_call = last_step[2]
    try:
        result = tool_executor.invoke(tool_call)
    except Exception as e:
        result = {"error": f"工具执行异常: {str(e)}"}
    return {
        "intermediate_steps": [("tool_result", tool_call["name"], result)],
        "retry_count": state.get("retry_count", 0) + 1
    }

def route_after_agent(state: AgentState) -> Literal["tools", "end"]:
    """条件路由：是否需要调用工具"""
    if state.get("final_output"):
        return "end"
    if state.get("intermediate_steps", []) and state["intermediate_steps"][-1][0] == "llm_decision":
        return "tools"
    return "end"

# 构建工作流图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", route_after_agent, {"tools": "tools", "end": END})
workflow.add_edge("tools", "agent")  # 工具执行后返回推理节点

app = workflow.compile()

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

def with_timeout(seconds: int = 10):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            try:
                return await asyncio.wait_for(func(*args, **kwargs), timeout=seconds)
            except asyncio.TimeoutError:
                return {"status": "timeout", "error": f"执行超时（>{seconds}s）"}
        return wrapper
    return decorator

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=5))
@with_timeout(15)
async def call_external_api(params: dict) -> dict:
    # 外部API调用实现
    pass

import resource
import subprocess
import tempfile
import os
import signal

def run_in_sandbox(code: str, timeout_sec: int = 5, memory_mb: int = 256) -> str:
    with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
        f.write(code)
        f.flush()
        script_path = f.name
    
    try:
        # 设置资源限制
        resource.setrlimit(resource.RLIMIT_CPU, (timeout_sec, timeout_sec + 1))
        resource.setrlimit(resource.RLIMIT_AS, (memory_mb * 1024 * 1024, memory_mb * 1024 * 1024))
        
        result = subprocess.run(
            ["python3", script_path],
            capture_output=True,
            text=True,
            timeout=timeout_sec,
            preexec_fn=lambda: signal.alarm(timeout_sec)
        )
        return result.stdout if result.returncode == 0 else result.stderr
    finally:
        os.unlink(script_path)

from aiolimiter import AsyncLimiter
import asyncio

# 每秒最多5次请求
rate_limiter = AsyncLimiter(max_rate=5, time_period=1)

async def rate_limited_invoke(func, *args, **kwargs):
    async with rate_limiter:
        return await func(*args, **kwargs)

from chromadb import PersistentClient
from sentence_transformers import SentenceTransformer
from typing import List

class VectorRetriever:
    def __init__(self, persist_path: str, collection_name: str, model_name: str = "BAAI/bge-large-zh-v1.5"):
        self.encoder = SentenceTransformer(model_name)
        self.client = PersistentClient(path=persist_path)
        self.collection = self.client.get_or_create_collection(collection_name)
    
    def retrieve(self, query: str, top_k: int = 3) -> List[str]:
        emb = self.encoder.encode(query).tolist()
        results = self.collection.query(query_embeddings=[emb], n_results=top_k)
        return results["documents"][0] if results else []

import logging
import time
from functools import wraps
from datetime import datetime

def trace_step(step_name: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(state: dict, *args, **kwargs):
            start = time.perf_counter()
            logging.info(json.dumps({
                "event": "step_start",
                "step": step_name,
                "input": state.get("user_input"),
                "timestamp": datetime.utcnow().isoformat()
            }))
            try:
                result = await func(state, *args, **kwargs)
                elapsed = time.perf_counter() - start
                logging.info(json.dumps({
                    "event": "step_end",
                    "step": step_name,
                    "elapsed_ms": elapsed * 1000,
                    "output_preview": str(result)[:200]
                }))
                return result
            except Exception as e:
                logging.error(f"Step {step_name} failed: {e}", exc_info=True)
                raise
        return wrapper
    return decorator

import tiktoken

def count_tokens(text: str, model: str = "gpt-4") -> int:
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

def should_compress(state: AgentState, max_tokens: int = 8000) -> bool:
    total = sum(count_tokens(msg["content"]) for msg in state["chat_history"])
    return total > max_tokens

# 子图示例
search_subgraph = create_search_agent_graph()
code_subgraph = create_code_agent_graph()

supervisor = StateGraph(OverallState)
supervisor.add_node("router", route_by_intent)
supervisor.add_node("search_worker", search_subgraph.compile())
supervisor.add_node("code_worker", code_subgraph.compile())
supervisor.add_conditional_edges("router", ...)

async def evaluate_agent(test_set: List[Tuple[str, str]]) -> float:
    total_score = 0.0
    for query, gold in test_set:
        result = await app.ainvoke({"user_input": query})
        score = await judge_with_llm(query, gold, result["final_output"])
        total_score += score
    return total_score / len(test_set)