企业官网建设流程全解析

1. 为什么“跑得起来”只是万里长征第一步？

“Agent 跑得起来，不等于管得住”——这句话不是危言耸听，而是我在过去两年深度参与多个开源智能体（Agent）项目后，被反复打脸、又反复验证的真实体会。它精准戳中了当前开源Agent生态里最普遍、也最容易被忽视的结构性断层：启动即胜利的幻觉，掩盖了持续运行的系统性风险。

你可能已经用LangChain搭出了一个能调用天气API、再写邮件总结的Demo；也可能用LlamaIndex+Ollama本地跑通了PDF问答流程；甚至在GitHub上一键clone了某个标着“Production Ready”的Agent框架，docker-compose up -d之后，终端跳出一连串绿色✅，日志里滚动着Agent initialized,Orchestrator started,Loop cycle completed……那一刻，你大概率会松一口气，觉得“成了”。

但问题恰恰始于这声“成了”。

我参与过一个高校实验室主导的跨平台文档协同Agent项目，初期版本上线时，所有功能链路都通了：用户上传PDF → Agent自动提取关键段落 → 调用本地大模型生成摘要 → 推送至企业微信通知群。整个流程在测试环境跑得飞快，响应时间平均2.3秒。团队庆祝完，第二天就收到第一波投诉：有3位用户反馈，他们上传的同一份合同文件，生成的摘要内容前后矛盾；还有人说，连续提交5次，第4次返回的是空结果，第5次却突然冒出一段完全无关的代码片段。更棘手的是，没人能立刻说清——到底是哪一次调用出的问题？是模型推理崩了？还是向量库检索错位？抑或是消息队列里某条指令被重复消费了？

这就是“管不住”的典型切口：可观测性缺失、状态不可追溯、异常无闭环。Agent不是单次执行的脚本，而是一个持续感知-决策-行动-反馈的闭环系统。它内部有状态（state）、有记忆（memory）、有工具调用链（tool call chain）、有外部依赖（API、数据库、消息中间件），还有不可控的LLM输出不确定性。当所有这些要素交织在一起，仅靠print("Step 3 done")或logger.info("Agent loop finished")这种粒度的日志，根本无法支撑任何有效诊断。

更值得警惕的是，很多开源Agent框架在设计之初，就把“可运行性”（runnability）当作核心KPI来优化，而把“可管理性”（manageability）默认交给使用者“自行补充”。比如，它们会花大力气封装一个优雅的AgentExecutor类，却对ExecutionTrace的结构定义极其简陋；会提供开箱即用的ReAct或Plan-and-Execute模板，却不内置任何用于回溯某次失败决策路径的上下文快照机制；会强调“支持多工具并行调用”，却对工具调用超时、重试、熔断、降级等生产级容错逻辑只字不提。

这背后是工程思维的错位：把Agent当成一个“增强版的函数调用”，而非一个需要全生命周期治理的“数字员工”。而现实是，一个真正要嵌入业务流程的Agent，其运维复杂度远超一个微服务——它没有固定的输入Schema（用户提问千奇百怪），没有确定的输出边界（LLM可能胡说八道），甚至没有稳定的执行时长（一次思考链可能耗时3秒，也可能卡死在某个工具调用上10分钟）。

所以，“跑得起来”只是完成了从0到1的验证，证明技术栈能拼接；而“管得住”，才是从1到N的门槛，它要求你建立一套与Agent行为深度耦合的治理体系：能看见它在想什么、做了什么、为什么这么做、哪里卡住了、如何安全地拉回来。这不是锦上添花的附加功能，而是决定Agent能否走出Demo、走进真实场景的生死线。

提示：不要被“Agent Framework”这个词迷惑。很多标榜“开箱即用”的框架，实际只提供了“执行引擎”这一层能力。真正的“框架”，必须包含可观测性、可调试性、可审计性、可干预性的完整支撑平面。如果它的文档里找不到trace_id、span、replay、rollback、sandbox这些词，那你拿到的很可能只是一个精美的执行器外壳。

2. 开源雷达周刊：我们到底在“看”什么？

“开源雷达周刊”这个名字，本身就暗示了一种主动扫描、动态感知、持续预警的运作逻辑。它不是一个静态的新闻聚合页，而是一套面向Agent生态的“健康监测仪表盘”。那么，这个“雷达”具体扫描哪些维度？它的“波束”应该聚焦在哪些关键信号上？这直接决定了你能否在问题爆发前，就捕捉到那些细微却危险的异常脉冲。

我把Agent的可观测性拆解为三个相互咬合的层次：执行层（Execution Layer）、决策层（Reasoning Layer）、治理层（Governance Layer）。开源雷达周刊的每期内容，本质上都是对这三个层次最新进展、工具、实践和坑的扫描与解读。

2.1 执行层：让每一次调用都“有迹可循”

这是最基础、也最容易被实现的层面，关注Agent作为一个程序实体，在物理世界中的运行痕迹。它回答的问题是：“它干了什么？”

• 调用链追踪（Distributed Tracing）：这是现代分布式系统的标配，但在Agent场景下有特殊挑战。一个典型的Agent调用链可能是：用户请求 → API网关 → Agent Orchestrator → 工具A（查数据库）→ 工具B（调外部API）→ 工具C（调用LLM）→ 结果聚合 → 响应返回。传统OpenTelemetry的Span结构，很难天然表达“工具C的调用是由工具B的返回结果触发的”这种强因果依赖。因此，像Langfuse、PromptLayer这类专注LLM可观测性的工具，开始引入parent_observation_id、trace_id、session_id等字段，强制建立跨工具、跨模型调用的上下文关联。开源雷达周刊会持续跟踪这类工具如何适配Agent的非线性执行流。

• 状态快照（State Snapshotting）：Agent的状态是其灵魂。一个ReAct Agent的状态，至少包含：当前步骤（step）、已执行动作列表（actions）、已获得观察结果（observations）、当前思考（thought）、下一步计划（plan）。每次循环迭代，都应该生成一个轻量级快照，并持久化到可查询的存储（如SQLite、PostgreSQL或专用向量数据库）。我见过太多项目，因为没做快照，导致线上故障复盘时，只能对着模糊的日志猜“当时它脑子里在想什么”。开源雷达周刊会评测不同快照方案的性能开销与恢复精度，比如基于JSON Schema的结构化快照 vs 基于内存dump的原始快照。

• 工具调用监控（Tool Call Monitoring）：这是执行层的“血压计”。你需要实时知道：每个工具的调用成功率、平均延迟、错误类型分布（网络超时？参数校验失败？下游服务503？）、重试次数。更重要的是，要能区分“工具本身失败”和“Agent错误地调用了不该调用的工具”。后者往往指向提示词（prompt）或规划（planning）模块的缺陷。开源雷达周刊会分析像langchain-community中BaseTool的invoke方法如何被Hook，以及如何在不侵入业务代码的前提下，注入统一的监控埋点。

2.2 决策层：解码“黑箱”里的思考链

如果说执行层是记录“做了什么”，那么决策层就是试图理解“为什么这么做”。这是Agent可观测性中最难、也最具价值的部分，它直指LLM的不确定性本质。

• 推理链（Chain-of-Thought, CoT）可视化：一个高质量的CoT日志，不应只是"I need to search for the latest news about AI..."这样的自然语言描述，而应是结构化的、可解析的。理想状态下，它应包含：reasoning_step_id、step_type（e.g.,query_generation,tool_selection,result_analysis）、input_context（该步所依据的上下文）、output_decision（最终做出的决策）、confidence_score（模型自评置信度，如果支持）。开源雷达周刊会追踪HuggingFace Transformers、vLLM等底层模型库，是否开始原生支持generate_with_reasoning_trace这类接口。

• 提示词（Prompt）版本与效果追踪：Agent的行为，70%以上由其提示词决定。但很多团队把提示词硬编码在Python文件里，或者只用一个prompt.txt文件管理。当效果变差时，根本无法回溯“是上周五更新的那个prompt版本导致了问题？还是模型升级带来的副作用？”开源雷达周刊会介绍promptfoo、promptlayer等工具如何将Prompt视为一等公民，进行版本控制、A/B测试、效果评估（通过预设的assertions或人工评分）。

• 幻觉（Hallucination）与事实性（Factuality）检测：这是决策层的“安检门”。不能只等用户投诉“你瞎说”，而要在Agent输出的瞬间，就启动轻量级的事实核查。开源雷达周刊会评测factcheck、hallucination-detection等开源库，它们如何利用检索增强（RAG）的原始知识源，或调用小型校验模型（如google/flan-t5-base），对Agent生成的每一句关键结论进行可信度打分。一个实用技巧是：对高风险领域（如医疗、法律、金融），强制设置factuality_threshold=0.95，低于此值则拒绝输出，转为“我需要更多信息来确认”。

2.3 治理层：赋予人类“叫停”与“修正”的权力

这是最高阶的层面，它超越了被动观测，进入了主动干预与策略制定。它回答的问题是：“当它做错了，我们能怎么办？”

• 沙盒（Sandbox）执行环境：这是最硬核的治理手段。任何高风险操作（如修改数据库、发送邮件、调用支付API），都必须先在一个隔离的、可回滚的沙盒环境中执行一遍。沙盒会模拟真实环境的所有依赖，但所有“写”操作都被重定向到临时存储或直接丢弃。只有当沙盒内执行成功且结果符合预期（通过预设规则或人工审核），才允许在真实环境中执行。开源雷达周刊会分析docker-sandbox、pysandbox等方案在Agent场景下的集成难度与性能损耗。

• 人工审核（Human-in-the-Loop, HITL）工作流：并非所有决策都适合全自动。对于涉及重大利益、伦理判断或模糊边界的请求，Agent应能主动“举手”请求人工介入。这需要一套标准化的HITL协议：Agent如何标记一个请求为“需审核”？审核界面如何展示完整的推理链和工具调用详情？审核员如何一键批准/拒绝/修改？审核结果如何反哺Agent的后续学习？开源雷达周刊会拆解langchain-hub中human_input_node的实战案例，以及如何用FastAPI快速搭建一个轻量级审核后台。

• 策略引擎（Policy Engine）：这是治理层的“大脑”。它是一套独立于Agent主逻辑的规则系统，负责执行全局策略。例如：“禁止在工作日22:00至次日6:00调用外部付费API”、“当连续3次工具调用失败，自动降级为仅使用本地缓存”、“对所有涉及‘赔偿’、‘诉讼’、‘违约’等关键词的输出，强制添加法律免责声明”。开源雷达周刊会对比jsonlogic、policy-engine等规则引擎，如何与主流Agent框架（如LangChain、LlamaIndex）进行低侵入式集成。

开源雷达周刊的价值，正在于它不满足于告诉你“某个新工具发布了”，而是深入到这些工具如何解决上述三个层次的具体问题，它们的适用边界在哪里，以及在真实项目中踩过哪些坑。它是一份给Agent工程师的“作战地图”，而不是一份给技术爱好者的“新品导购”。

3. “管不住”的五大典型症状与根因定位

“管不住”不是一种抽象感受，它会以非常具体、甚至令人抓狂的症状表现出来。识别这些症状，是走向“管得住”的第一步。下面这五种情况，我在不同项目中反复遇到，每一次都伴随着数小时甚至数天的排查。我把它们称为“Agent失治五症”，并附上我的根因定位路径——这不是教科书式的标准答案，而是从血泪经验中提炼出的、可立即上手的排查清单。

3.1 症状一：结果“时好时坏”，无法稳定复现

现象：同一个用户输入，在一分钟内连续提交3次，第一次返回正确答案，第二次返回空，第三次返回完全无关的内容。日志里没有任何ERROR或WARNING，所有调用链都显示“completed”。

根因定位路径：

1. 首先排除LLM随机性：检查是否启用了temperature=0。如果temperature > 0，这是LLM的固有特性，不是Bug，而是Feature。解决方案是：在生产环境，对关键任务强制temperature=0，并接受其可能牺牲部分创造性。
2. 检查状态污染：这是最常见的元凶。Agent的memory（如ConversationBufferMemory）是否在多次请求间被意外共享？一个典型的错误是：将memory实例作为全局变量或类属性，而非每次请求都新建一个。用print(id(memory))在每次请求入口处打印内存对象ID，如果ID相同，则100%是状态污染。
3. 检查工具调用的“副作用”：某些工具（尤其是自定义工具）在执行时，可能会修改全局变量、缓存或单例对象。例如，一个数据库查询工具，在查询后顺手把结果存到了一个全局CACHE_DICT里，而下一次调用时，它直接从这个被污染的缓存里读取了旧数据。解决方案：为每个工具调用创建一个干净的、隔离的执行上下文（context）。
4. 检查向量库检索的“漂移”：如果你的Agent严重依赖RAG，那么向量库的索引是否在后台被其他进程更新？检索时使用的embedding模型版本，是否与索引构建时的版本一致？一个简单的验证方法：固定一个查询，手动用相同的embedding模型去计算其向量，然后用faiss或chromadb的search接口直接查询，看结果是否稳定。

注意：当遇到“时好时坏”，永远优先怀疑“状态”和“共享资源”，而不是第一时间去骂LLM。LLM的输出是确定性的（给定相同输入、相同参数、相同模型权重），所有不确定性都来自外部。

3.2 症状二：执行“无限循环”，CPU飙升至100%

现象：Agent开始处理一个请求后，不再返回任何结果，服务器CPU持续满载，日志里疯狂刷屏Executing step 5... Executing step 5... Executing step 5...。

根因定位路径：

1. 检查终止条件（Stop Condition）：这是最直接的原因。Agent的循环逻辑里，是否有一个明确、可靠的退出条件？例如，if "FINAL ANSWER:" in output or step_count > MAX_STEPS:。很多开源Demo为了简化，直接用while True:，然后靠break，但break的条件可能永远不满足。解决方案：在循环入口处，强制加入step_count += 1; if step_count > 10: raise RuntimeError("Max steps exceeded")。
2. 检查工具调用的“死锁”：Agent规划了一个动作，但该动作对应的工具，因为网络超时、下游服务无响应或参数错误，一直卡在await或time.sleep()上，导致Agent主循环无法推进。解决方案：为所有工具调用设置严格的timeout（如asyncio.wait_for(tool.invoke(), timeout=15)），并在超时后抛出特定异常，由Agent的错误处理逻辑捕获并进入降级流程。
3. 检查LLM输出的“格式错误”：Agent期望LLM输出一个严格遵循Action: Search\nAction Input: "AI news"的格式，但LLM偶尔会输出Action: Search\nAction Input: ["AI news"]（加了方括号）或Action: Search\nAction Input: "AI news"\nThought: ...（多了Thought）。这会导致Agent的解析器（parser）崩溃或进入无效分支，从而无法生成有效的下一步动作。解决方案：在Parser层增加鲁棒性，例如用正则表达式r"Action:\s*(\w+)\s*Action Input:\s*(.+)"，并设置最大重试次数。

3.3 症状三：工具“调用成功”，但结果“完全不对”

现象：日志显示Tool 'Search' invoked successfully with input 'latest AI conference'，但Agent后续的思考链却基于一个完全错误的搜索结果（比如返回了2018年的旧闻）。

根因定位路径：

1. 检查工具本身的逻辑：这是最常被忽略的。不要假设工具是“正确的”。用Postman或curl，直接调用该工具封装的底层API，传入完全相同的参数，看返回结果是否真的如日志所言。我曾在一个项目中发现，一个名为WebSearch的工具，其内部实现竟然是一个硬编码的return "Here is some fake search result"，只为方便本地测试，结果被误提交到了生产环境。
2. 检查工具输入的“语义漂移”：Agent规划的Action Input，是否真的表达了用户的意图？例如，用户问“帮我找一下昨天发布的关于Stable Diffusion 3的新闻”，Agent可能规划出Action Input: "Stable Diffusion 3 news"。这个输入丢失了最关键的“时间”约束。解决方案：在工具调用前，增加一个“输入增强”（Input Augmentation）步骤，利用LLM将原始规划输入，重写为更精确、更符合工具API要求的查询语句。
3. 检查RAG检索的“相关性陷阱”：如果工具是基于向量库的检索，那么“成功”只意味着向量相似度高，不代表语义相关。一个关于“Apple”的搜索，可能因为“苹果公司”和“苹果手机”的向量相近，而返回一堆关于iPhone的文档。解决方案：在检索后，增加一个“重排序”（Rerank）步骤，使用Cross-Encoder模型（如BAAI/bge-reranker-base）对Top-K结果进行二次打分和排序。

3.4 症状四：日志“信息过载”，却“找不到关键线索”

现象：日志文件每天增长几个G，里面充斥着DEBUG级别的海量细节，但当你想查“用户ID为12345的这次请求，为什么返回了错误答案？”时，却像大海捞针，需要手动grep几十个关键词，再拼凑线索。

根因定位路径：

1. 强制引入trace_id：这是所有现代可观测性的基石。在用户请求进入系统的第一个入口（如FastAPI的@app.post("/chat")），就生成一个唯一的trace_id（如uuid.uuid4().hex[:8]），并将其作为logging.Logger的extra参数，贯穿整个请求生命周期。所有日志，无论来自Agent、工具、还是数据库驱动，都必须带上这个trace_id。这样，你只需要grep "trace_id=abc123de" app.log，就能拿到该次请求的全部日志流。
2. 结构化日志（Structured Logging）：抛弃logger.info(f"User {user_id} asked {query}")这种字符串拼接。改用logger.info("User asked question", user_id=user_id, query=query, trace_id=trace_id)。这样，日志会被序列化为JSON，可以被ELK（Elasticsearch, Logstash, Kibana）或Loki+Grafana等专业日志系统轻松索引和查询。
3. 定义关键事件（Key Events）：不要记录一切，而要定义什么是“关键”。例如，AGENT_START,TOOL_INVOKE_START,TOOL_INVOKE_END,LLM_GENERATE_START,LLM_GENERATE_END,AGENT_FINISH。为每个关键事件打上status（success/fail）、duration_ms、error_message（如果失败）等字段。这样，你一眼就能看出，是哪个环节拖慢了整体速度，或是哪个环节频繁失败。

3.5 症状五：上线后“一切正常”，但“用户悄悄流失”

现象：监控面板上，QPS、成功率、平均延迟等所有指标都绿油油的，但业务侧反馈，用户咨询量在下降，或者用户在对话中越来越倾向于说“算了，我自己查吧”。

根因定位路径：

1. 建立“体验指标”（Experience Metrics）：这是比“技术指标”更难，但也更重要的维度。你需要主动测量：
   • 首次响应时间（First Response Time）：用户发出第一条消息，到收到Agent第一条回复的时间。超过5秒，用户耐心就会急剧下降。
   • 对话轮次（Turns per Session）：一个有效对话，平均需要多少轮交互？如果这个数字在上升，说明Agent的单次回答质量在下降，用户需要反复追问。
   • 人工接管率（Handoff Rate）：Agent主动请求人工审核，或用户主动点击“转人工”按钮的比例。这是一个极强的负向信号。
2. 部署“影子模式”（Shadow Mode）：在生产环境，让新版本的Agent与旧版本并行运行。新版本的输出不返回给用户，只用于收集其决策链、工具调用、输出内容，并与旧版本的输出进行自动化对比（例如，用BLEU或ROUGE分数衡量文本相似度，或用规则匹配关键信息点）。这样，你可以在不打扰用户的情况下，提前发现新版本的潜在劣化。
3. 进行“用户反馈闭环”：在Agent的每次回复末尾，加上一个极简的反馈按钮：“👍 有帮助 / 👎 没帮助”。收集到的反馈，必须实时关联到该次请求的trace_id，并进入一个专门的分析看板。不要只看总数，要分析“没帮助”的反馈，是否集中在某个特定的工具调用（如Search）或某个特定的LLM输出模式（如总是过度解释）。

这五种症状，就像人体的五种“病症”，它们指向的是Agent系统不同器官的“亚健康”状态。识别它们，不是为了指责某个组件，而是为了精准地找到那个需要被“体检”和“调理”的部位。开源雷达周刊的核心价值之一，就是持续为你提供针对这些“病症”的最新“诊疗方案”和“预防指南”。

4. 构建你的第一套Agent治理流水线：从零到一的实操指南

理论讲得再多，不如亲手搭起一条能跑起来的治理流水线。下面，我将带你用最精简、最务实的技术栈，构建一个具备基础可观测性、可调试性和可干预能力的Agent治理流水线。它不追求大而全，而是确保每一个环节都“看得见、摸得着、管得了”。整个过程，我会用一个真实的、可立即运行的代码示例来贯穿。

4.1 技术选型：为什么是这套组合？

在动手之前，必须明确选择理由。这不是一个随意的堆砌，而是基于“最小可行治理”的原则，对每个组件的权衡：

• Agent框架：LangChain
  理由：生态最成熟，文档最完善，社区最活跃，且其CallbackHandler机制是构建可观测性的绝佳入口。虽然它不是最轻量的，但对于初学者和快速验证来说，它的“开箱即用”优势无可替代。我们不会用它所有的高级特性，只聚焦于Runnable、Tool和CallbackHandler这三个核心。

• 可观测性后端：Langfuse
  理由：它是目前唯一一个专为LLM应用设计的、开源且自托管的可观测性平台。它原生支持trace、span、generation、observation等概念，完美契合Agent的执行模型。相比Prometheus+Grafana（更适合基础设施监控）或ELK（更适合通用日志），Langfuse能让你在5分钟内，就看到一个完整的Agent调用链图谱。它还免费提供云托管版，可以零成本快速上手。

• 沙盒执行：Pythonsubprocess+timeout
  理由：不引入任何第三方沙盒库（如pysandbox，它过于重量级且维护不佳）。对于绝大多数Agent工具（如调用curl、python脚本、或本地CLI工具），subprocess.run(cmd, timeout=30, capture_output=True)配合严格的timeout和check=True，就能提供足够安全的隔离。它简单、透明、可控，且无需额外部署。

• 人工审核：一个50行的FastAPI后台
  理由：避免陷入复杂的审批流引擎。一个极简的Web界面，能展示trace_id、原始输入、Agent的完整思考链、所有工具调用详情、以及最终输出，就足以支撑初期的人工审核需求。FastAPI的@app.get("/review/{trace_id}")路由，配合一个Jinja2模板，50行代码就能搞定。

这套组合的总学习成本，远低于去研究一个全新的、小众的Agent框架。它的目标是：让你在2小时内，就拥有一套能真正解决问题的治理能力，而不是在选型上纠结两周。

4.2 核心代码：一个可运行的治理Agent

下面是一个完整的、可直接复制粘贴运行的Python脚本。它实现了一个具备基础治理能力的Agent，其核心在于CustomCallbackHandler类，它是我们整个治理流水线的“神经中枢”。

# agent_with_governance.py import asyncio import json import logging import subprocess import time from typing import Any, Dict, List, Optional, Union from uuid import uuid4 from langchain.callbacks.base import BaseCallbackHandler from langchain.chains import LLMChain from langchain.chat_models import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.schema import HumanMessage, SystemMessage from langchain.tools import BaseTool from langchain.tools.render import format_tool_to_openai_function from langchain.utils.openai_functions import convert_pydantic_to_openai_function from pydantic import BaseModel, Field # 配置日志，确保trace_id能被所有日志携带 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class CustomCallbackHandler(BaseCallbackHandler): """自定义回调处理器，是治理流水线的核心""" def __init__(self, trace_id: str): self.trace_id = trace_id # 这里可以初始化连接到Langfuse等后端 # 为简化，我们先用内存存储，模拟上报 self.spans = [] self.start_time = time.time() def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any) -> None: """Agent主链路开始""" span = { "id": f"span_{uuid4().hex[:6]}", "name": "Agent Main Loop", "type": "chain", "start_time": time.time(), "inputs": {"query": inputs.get("input", "N/A")}, "trace_id": self.trace_id, } self.spans.append(span) logger.info(f"[TRACE:{self.trace_id}] Agent started with input: {inputs.get('input', 'N/A')}") def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) -> None: """工具调用开始""" tool_name = serialized.get("name", "unknown") span = { "id": f"span_{uuid4().hex[:6]}", "name": f"Tool: {tool_name}", "type": "tool", "start_time": time.time(), "inputs": {"input": input_str}, "trace_id": self.trace_id, } self.spans.append(span) logger.info(f"[TRACE:{self.trace_id}] Tool '{tool_name}' invoked with input: {input_str}") def on_tool_end(self, output: str, **kwargs: Any) -> None: """工具调用结束""" # 找到最近的tool span for span in reversed(self.spans): if span["type"] == "tool" and "end_time" not in span: span["end_time"] = time.time() span["outputs"] = {"output": output} span["duration_ms"] = (span["end_time"] - span["start_time"]) * 1000 logger.info(f"[TRACE:{self.trace_id}] Tool completed in {span['duration_ms']:.0f}ms. Output: {output[:100]}...") break def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None: """Agent主链路结束""" for span in self.spans: if span["name"] == "Agent Main Loop" and "end_time" not in span: span["end_time"] = time.time() span["outputs"] = outputs span["duration_ms"] = (span["end_time"] - span["start_time"]) * 1000 logger.info(f"[TRACE:{self.trace_id}] Agent finished in {span['duration_ms']:.0f}ms. Final answer: {outputs.get('output', 'N/A')[:100]}...") break # 【关键治理点】此处可模拟上报到Langfuse self._report_to_observability_backend() def _report_to_observability_backend(self): """模拟上报到可观测性后端""" # 在真实项目中，这里会调用Langfuse SDK # langfuse.trace( # name="agent_run", # id=self.trace_id, # input={"spans": self.spans}, # output={"final_answer": self.spans[-1].get("outputs", {}).get("output", "")} # ) print(f"\n--- TRACE REPORT FOR {self.trace_id} ---") for span in self.spans: duration = span.get("duration_ms", "N/A") print(f" {span['name']} | {duration}ms | Status: {'Completed' if 'end_time' in span else 'Running'}") print("--- END REPORT ---\n") # 定义一个“沙盒化”的搜索工具 class SandboxSearchTool(BaseTool): name = "Search" description = "A sandboxed web search tool. It runs in a separate process with strict timeout." def _run(self, query: str) -> str: """在沙盒中执行搜索""" try: # 模拟一个可能很慢或失败的外部命令 # 在真实场景中，这里会是 `curl` 或 `python search_script.py` result = subprocess.run( ["echo", f"Simulated search results for: {query}"], capture_output=True, text=True, timeout=5, # 严格5秒超时 check=True ) return result.stdout.strip() except subprocess.TimeoutExpired: return "ERROR: Search timed out after 5 seconds." except subprocess.CalledProcessError as e: return f"ERROR: Search failed with exit code {e.returncode}." except Exception as e: return f"ERROR: Unexpected error in search: {str(e)}" async def _arun(self, query: str) -> str: # 异步版本，原理相同 return self._run(query) # 创建Agent def create_governed_agent(): # 初始化LLM（这里用mock，实际可替换为OpenAI或Ollama） llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0) # 定义工具 tools = [SandboxSearchTool()] # 构建提示词模板（简化版ReAct） prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant. Use the provided tools to answer the question. Think step by step."), ("user", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), ]) # 创建Agent（这里用LangChain的`create_react_agent`，但你可以用任何方式） # 为简化，我们手动构建一个极简的循环 from langchain.agents.format_scratchpad import format_to_openai_function_messages from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser # 【关键治理点】为Agent注入回调处理器 trace_id = f"trc_{uuid4().hex[:8]}" callback_handler = CustomCallbackHandler(trace_id=trace_id) # 这里省略了完整的Agent构建代码，重点是：所有LLM调用和Tool调用，都必须经过callback_handler # 在真实项目中，你会将callback_handler传入`llm`和`tool`的构造函数 return lambda query: _execute_agent_loop(query, llm, tools, callback_handler, trace_id) def _execute_agent_loop(query: str, llm, tools, callback_handler, trace_id): """模拟一个极简的Agent执行循环""" # 1. 记录Agent启动 callback_handler.on_chain_start({"name": "GovernedAgent"}, {"input": query}) # 2. 模拟一次工具调用 tool = tools[0] callback_handler.on_tool_start({"name": tool.name}, query) tool_result = tool._run(query) callback_handler.on_tool_end(tool_result) # 3. 模拟LLM生成最终答案 # 这里用一个mock的LLM响应 final_answer = f"Based on the search results, here's what I found: {tool_result}" # 4. 记录Agent结束 callback_handler.on_chain_end({"output": final_answer}) return final_answer # 主程序入口 if __name__ == "__main__": # 创建受治理的Agent governed_agent = create_governed_agent() # 测试运行 test_query = "What's the latest news about open-source AI?" print(f"Running query: {test_query}") result = governed_agent(test_query) print(f"Final Result: {result}")

4.3 运行与验证：亲眼看到“治理”发生

1. 安装依赖：pip install langchain openai python-dotenv
2. 设置环境变量：创建.env文件，填入你的OpenAI API Key（如果要用真实模型）。
3. 运行脚本：python agent_with_governance.py
4. 观察输出：你会看到