企业官网建设流程全解析

简介：本资源是一份面向AI智能体开发者与软件工程师的轻量级源码对比实践包，聚焦ReAct（推理-行动）与Plan-and-Execute（规划-执行）两大核心运行模式的技术选型与落地差异。资源通过可运行的HTML示例、.inscode配置说明及标准Git工程结构，直观呈现两种模式在代码组织、控制流设计与环境响应逻辑上的本质区别，助力开发者理解何时选用、如何切换或融合二者以提升智能体鲁棒性。压缩包共3个文件（10KB），含HTML前端演示页（支持本地直接打开查看交互逻辑）、.inscode配置文件（定义智能体行为元信息）及.gitignore（体现工程化规范），结构精简、即开即用。目前已有58人学习下载，适合中初级AI工程实践者快速掌握模式差异、复现关键流程并嵌入自身项目架构设计。

1. Agent运行模式对比：不是选“快”或“省”，而是看任务闭环卡在哪一环

你写好了一个Agent，本地跑通了天气查询+日程提醒的链路，但一上生产环境就掉链子：用户问“帮我订明天下午三点前最便宜的高铁票”，它先查了天气、又翻出上周会议纪要、最后返回“暂不支持订票”——不是模型不会推理，是运行模式没对齐任务本质。
“Agent运行模式对比[源码]”这个标题，拆开看其实是三个硬核问题：任务如何被切片？状态如何被携带？失败如何被兜底？它不讲LLM多强，而聚焦在“指令下发后，系统怎么一步步把事办成”。适合正在从单步Function Call转向多跳任务编排的开发者，尤其当你发现：调用链越来越长、超时频发、中间结果无法复用、debug时像在黑匣子外敲门——那不是prompt没写好，是运行模式选错了。本文不讲抽象范式，只拿真实可跑的源码结构说话：ReAct、Plan-and-Execute、Toolformer-style Streaming、Stateful Loop 四种主流模式，每种都给出最小可验证实现、关键参数含义、以及我在线上踩出的血泪坑。

2. 四种Agent运行模式的源码骨架与核心差异

提示：本节所有代码均基于 Python 3.10+ +langchain-core==0.3.0+pydantic==2.9.2构建，不依赖任何闭源SDK或云服务。所有模式均可在单机CPU环境跑通，无需GPU。

2.1 ReAct模式：用“思考-行动-观察”三步闭环对抗幻觉

ReAct（Reasoning + Acting）不是简单加个Thought:前缀，而是强制将推理与工具调用解耦为原子步骤。其源码骨架的关键在于状态不可跨步共享：每一步的observation必须显式传入下一步，不能靠全局变量缓存。

# react_mode.py from typing import List, Dict, Any from pydantic import BaseModel class ReActStep(BaseModel): thought: str action: str # 工具名，如 "search_web" action_input: str # 工具参数，如 "上海天气" observation: str = "" def run_react_loop( initial_prompt: str, tools: Dict[str, callable], max_steps: int = 6 ) -> List[ReActStep]: steps = [] current_input = initial_prompt for step_idx in range(max_steps): # Step 1: LLM生成Thought+Action（固定prompt模板） llm_output = call_llm_with_template( prompt=f"Question: {current_input}\nThought:", stop=["\nObservation:"] ) # 实际中用openai.ChatCompletion或本地llm thought_action = parse_thought_action(llm_output) # Step 2: 执行Action（严格校验工具名和参数） if thought_action.action not in tools: raise ValueError(f"Unknown tool: {thought_action.action}") try: observation = tools[thought_action.action](thought_action.action_input) except Exception as e: observation = f"ERROR: {str(e)}" # Step 3: 记录当前步，observation作为下一轮输入的一部分 step = ReActStep( thought=thought_action.thought, action=thought_action.action, action_input=thought_action.action_input, observation=observation ) steps.append(step) # 终止条件：LLM明确输出Answer if "Answer:" in observation or "Final Answer:" in observation: break # 下一轮输入 = 原始问题 + 所有历史step（压缩版） current_input = f"Question: {initial_prompt}\n" + \ "\n".join([f"Thought: {s.thought}\nAction: {s.action}\nAction Input: {s.action_input}\nObservation: {s.observation}" for s in steps]) return steps

参数说明与调优逻辑：

• max_steps：不是越大越好。线上实测超过8步后，LLM对长上下文的理解准确率断崖下降（某高校模拟项目X中，6步时任务完成率82%，10步跌至41%）。建议从4起步，按任务复杂度阶梯增加。
• stop=["\nObservation:"]：必须显式设stop token，否则LLM可能生成冗余文本导致parse_thought_action解析失败。这是ReAct模式最易翻车的点——别信“模型会自动停”。
• current_input拼接逻辑：绝不拼接原始observation全文。实测中，将10KB的网页HTML直接塞进prompt会导致token爆炸且LLM注意力偏移。正确做法是：用轻量摘要函数（如summarize_observation(obs)）压缩到200字内再拼接。

2.2 Plan-and-Execute模式：用静态计划表规避动态决策噪声

当任务路径高度确定（如“查航班→比价格→填乘客→支付”），Plan-and-Execute 比ReAct更稳。它的核心是两阶段分离：第一阶段LLM输出JSON格式的执行计划（Plan），第二阶段按计划顺序执行（Execute），不回传中间结果给LLM。

# plan_execute_mode.py import json from pydantic import BaseModel class PlanStep(BaseModel): step_id: int tool: str input: str expected_output: str # LLM预判该步应产出什么，用于后续校验 def generate_plan(prompt: str, tools: List[str]) -> List[PlanStep]: # LLM仅输出JSON数组，无任何额外文本 plan_json_str = call_llm_with_template( prompt=f"Generate a step-by-step plan to answer: '{prompt}'. " f"Available tools: {tools}. Output ONLY valid JSON array.", response_format={"type": "json_object"} # OpenAI 1.0+ 支持 ) return [PlanStep(**item) for item in json.loads(plan_json_str)] def execute_plan(plan: List[PlanStep], tools: Dict[str, callable]) -> Dict[str, Any]: results = {} for step in plan: try: # 关键：输入仅来自上一步output或原始prompt，不引入LLM新推理 actual_input = step.input if "{" in step.input and "}" in step.input: # 含变量占位符 actual_input = step.input.format(**results) result = tools[step.tool](actual_input) results[f"step_{step.step_id}"] = result # 可选：校验LLM预判与实际结果是否匹配（防工具异常） if not fuzzy_match(result, step.expected_output): print(f"Warning: step {step.step_id} output mismatch") except Exception as e: results[f"step_{step.step_id}"] = f"EXEC_ERROR: {e}" break # 严格失败即停，不重试 return results

为什么它抗干扰？

• Plan阶段LLM只做“路径规划”，不接触实时数据，避免因工具返回噪声（如API限流提示）导致推理偏航。
• Execute阶段完全去LLM化，纯函数调用，性能稳定、可观测性强。某跨平台系统上线后，P95延迟从1.2s降至320ms。
• expected_output字段是后悔药：当工具返回异常值（如航班查询返回空列表），可触发降级策略（如切换备用API），而非让LLM强行解释。

2.3 Toolformer-style Streaming模式：用流式Token降低首字延迟

当用户需要“边想边说”的交互感（如客服对话中实时反馈“正在查询订单…”），传统Agent的“等整轮输出完再执行”太反直觉。Toolformer-style Streaming 的核心是LLM输出Token流中嵌入工具调用指令，实现毫秒级响应。

# streaming_mode.py import asyncio from typing import AsyncGenerator async def stream_with_tools( prompt: str, tools: Dict[str, callable], chunk_callback: callable # 接收每个token的回调函数 ) -> AsyncGenerator[str, None]: # Step 1: 启动LLM流式生成 async for token in call_llm_stream(prompt): await chunk_callback(token) # 立即推给前端 yield token # Step 2: 实时检测特殊标记（如<tool:search>query</tool>） if is_tool_marker(token): tool_name, tool_input = extract_tool_from_token(token) if tool_name in tools: try: # 异步执行工具，不阻塞token流 result = await asyncio.to_thread(tools[tool_name], tool_input) # 将结果作为“系统消息”注入流，带特殊前缀 yield f"\n<system_result>{result}</system_result>\n" except Exception as e: yield f"\n<system_error>{e}</system_error>\n"

关键约束与取舍：

• is_tool_marker()必须极轻量：正则匹配<tool:(\w+)>(.*?)</tool>，不能做JSON解析。线上压测显示，每token做json.loads()会使TPS下降67%。
• chunk_callback是性能瓶颈点：若回调里含HTTP请求，会拖垮整个流。正确做法是回调只写入内存队列，由独立线程消费。
• 不适用于强一致性场景：因为工具结果以<system_result>形式插入流，前端需自行解析并渲染，无法保证“思考-行动-结果”严格按序呈现。适合信息型任务（查天气、搜文档），慎用于事务型（下单、转账）。

2.4 Stateful Loop模式：用持久化状态支撑长周期任务

当任务跨度数小时甚至数天（如“监控服务器CPU>90%持续5分钟，自动扩容并通知负责人”），传统模式无法维持上下文。Stateful Loop 的核心是将Agent状态存入外部存储，并在每次唤醒时加载。

# stateful_mode.py import sqlite3 from datetime import datetime class AgentState(BaseModel): session_id: str last_step_time: datetime context: Dict[str, Any] # 如 {"cpu_history": [92,88,95,...], "alert_sent": False} next_action: str # 下次该执行什么，如 "check_cpu" def get_state(session_id: str) -> AgentState: conn = sqlite3.connect("agent_states.db") cursor = conn.cursor() cursor.execute("SELECT state_json FROM states WHERE session_id = ?", (session_id,)) row = cursor.fetchone() conn.close() if row: return AgentState.model_validate_json(row[0]) return AgentState(session_id=session_id, last_step_time=datetime.now(), context={}, next_action="init") def save_state(state: AgentState): conn = sqlite3.connect("agent_states.db") cursor = conn.cursor() cursor.execute( "INSERT OR REPLACE INTO states (session_id, state_json, updated_at) VALUES (?, ?, ?)", (state.session_id, state.model_dump_json(), datetime.now()) ) conn.commit() conn.close() def run_stateful_loop(session_id: str, tools: Dict[str, callable]): state = get_state(session_id) # 根据next_action分发执行 if state.next_action == "check_cpu": cpu_usage = tools["get_cpu_usage"]() state.context.setdefault("cpu_history", []).append(cpu_usage) state.context["cpu_history"] = state.context["cpu_history"][-5:] # 只存最近5次 if len(state.context["cpu_history"]) >= 5 and all(x > 90 for x in state.context["cpu_history"]): tools["scale_up_instance"]() tools["send_alert"]("CPU overload detected!") state.next_action = "done" else: state.next_action = "wait_60s" elif state.next_action == "wait_60s": # 模拟等待，实际中用定时任务触发 state.next_action = "check_cpu" state.last_step_time = datetime.now() save_state(state)

存储选型血泪经验：

• SQLite 足够应付QPS<50的场景，但必须加WAL模式：PRAGMA journal_mode=WAL;，否则并发写入时锁表导致超时。
• Redis 更适合高频状态更新（如每秒检查），但需自行处理过期逻辑（EXPIRE）和序列化（推荐msgpack而非JSON，体积小40%）。
• 绝对禁止用内存变量存state：进程重启即丢失，线上事故高发区。某实验室曾因未持久化监控状态，导致凌晨扩容失败无人知晓。

3. 运行模式避坑指南：5条线上真·翻车记录

3.1 ReAct模式：LLM在第3步突然开始“自我辩论”，无限循环

现象：Agent执行到第3步，LLM输出Thought: I need to verify the weather data... Action: search_web Action Input: "Shanghai weather verification"，但search_web返回的是原始天气数据，LLM却再次要求搜索“verification”，形成死循环。
原因：current_input拼接时未过滤重复步骤。当observation含“2024年7月15日上海多云”时，LLM误以为这是新问题而非历史结果，触发冗余验证。
解决：在拼接current_input前，对observation做指纹去重——计算MD5前8位，若与历史step的指纹相同，则跳过该step的拼接。代码加在run_react_loop的拼接逻辑前：

# 在拼接current_input前插入 obs_fingerprint = hashlib.md5(step.observation.encode()).hexdigest()[:8] if obs_fingerprint not in seen_fingerprints: seen_fingerprints.add(obs_fingerprint) # 再拼接此step

3.2 Plan-and-Execute模式：Plan阶段输出JSON格式错误，程序直接崩溃

现象：LLM偶尔在Plan阶段输出{"steps": [...]}（顶层是对象）或[{"step_id":1}, ...]（数组），但代码强制要求List[PlanStep]，json.loads()后类型校验失败。
原因：LLM对response_format={"type": "json_object"}的支持不稳定，尤其在开源模型上。OpenAI 1.0+虽宣称支持，但实测仍有3.2%概率返回非标准JSON。
解决：加一层容错解析器，不依赖json.loads()直抛异常：

def robust_parse_plan(json_str: str) -> List[PlanStep]: # 先尝试标准JSON try: data = json.loads(json_str) if isinstance(data, list): return [PlanStep(**item) for item in data] elif isinstance(data, dict) and "steps" in data: return [PlanStep(**item) for item in data["steps"]] except: pass # 再尝试正则提取关键字段（保底方案） step_ids = re.findall(r'"step_id"\s*:\s*(\d+)', json_str) tools = re.findall(r'"tool"\s*:\s*"([^"]+)"', json_str) inputs = re.findall(r'"input"\s*:\s*"([^"]+)"', json_str) if step_ids and tools and inputs: return [PlanStep(step_id=int(i), tool=t, input=inpt) for i,t,inpt in zip(step_ids, tools, inputs)] raise ValueError("Cannot parse plan from LLM output")

3.3 Streaming模式：工具执行耗时2秒，但LLM流已结束，用户看到空白结果

现象：用户提问后，前端快速刷出“正在查询...”，但2秒后工具结果返回时，LLM流早已关闭，<system_result>被丢弃。
原因：call_llm_stream()返回AsyncGenerator，但未与工具执行协程做生命周期绑定。LLM流结束即释放资源，不管工具是否完成。
解决：用asyncio.wait()同步LLM流与工具执行，确保两者都完成才退出：

async def stream_with_tools_safe(...): llm_task = asyncio.create_task(aiter_to_list(call_llm_stream(prompt))) # 收集所有token tool_task = None async for token in call_llm_stream(prompt): yield token if is_tool_marker(token): tool_task = asyncio.create_task(execute_tool_async(...)) # 等待LLM流结束 AND 工具执行完成 await llm_task if tool_task: await tool_task

3.4 Stateful Loop模式：SQLite数据库被多个进程同时写入，报错database is locked

现象：高并发场景下，大量Agent session同时调用save_state()，SQLite报OperationalError: database is locked。
原因：默认SQLite写操作是串行的，INSERT OR REPLACE会锁整张表。
解决：

1. 启用WAL模式（启动时执行PRAGMA journal_mode=WAL;）；
2. 在save_state()中加重试逻辑，指数退避：

def save_state_with_retry(state: AgentState, max_retries=3): for i in range(max_retries): try: # 原save_state逻辑 return except sqlite3.OperationalError as e: if "database is locked" in str(e) and i < max_retries - 1: time.sleep(0.1 * (2 ** i)) # 0.1s, 0.2s, 0.4s continue raise

3.5 四种模式共性坑：工具函数抛出异常时，未记录原始traceback，debug全靠猜

现象：tools["search_web"]("error query")抛出requests.Timeout，但Agent只返回"EXEC_ERROR: timeout"，无法定位是网络问题还是DNS问题。
原因：异常捕获过于宽泛，str(e)丢失了关键上下文。
解决：统一用traceback.format_exc()记录完整栈：

except Exception as e: full_trace = traceback.format_exc() results[f"step_{step.step_id}"] = f"EXEC_ERROR: {e}\n{full_trace[:500]}" # 截断防爆token

4. 如何选择你的Agent运行模式：一张决策表与三个验证技巧

选模式不是看文档多炫酷，而是看它能否扛住你的真实流量。以下是我在某图像处理Demo和某跨平台系统中沉淀的决策表，覆盖90%常见场景：

4.1 验证技巧一：用“故障注入法”测模式韧性

不要等线上出事才验证。在测试环境主动注入三类故障：

• 工具超时：用unittest.mock.patch将工具函数替换为lambda x: time.sleep(3) or "mock_result"，观察Agent是否降级（如ReAct是否跳过该步，Stateful是否重试）。
• LLM乱输出：在prompt中插入<random_junk>abc123</random_junk>，验证parse_thought_action()能否安全忽略。
• 状态损坏：手动篡改SQLite中某条state_json为非法JSON，确认get_state()有兜底逻辑（如返回默认state）。

4.2 验证技巧二：监控“决策熵”识别模式失配

LLM的每一步输出都有不确定性。我们定义决策熵：对LLM输出的action字段，统计其在历史100次中的分布方差。

• 低熵（方差<0.1）：LLM总选同一工具，说明任务简单，Plan-and-Execute足够；
• 高熵（方差>0.5）：LLM在多个工具间摇摆，ReAct或Streaming更合适；
• 熵值突变：某天起熵从0.2飙升至0.8，大概率是工具API变更（如搜索接口返回字段调整），触发模式切换预警。

4.3 验证技巧三：用“人工接管点”检验可维护性

在Agent代码中埋3个硬编码断点：

1. Plan生成后，打印plan_json并input("Press Enter to execute...")；
2. 每步observation返回后，打印step.thought并暂停；
3. Stateful Loop每次save_state()前，打印state.context。
   如果开发同学能在5分钟内通过这些断点定位到问题（如发现observation含HTML标签导致LLM误判），说明模式设计是可维护的；如果断点后仍需翻1000行日志，那模式本身就有问题。

我坚持在每个新Agent项目启动时，用这三招跑一遍——不是为了证明模式多完美，而是确保当凌晨三点告警响起时，我能5分钟内判断是“模式选错了”，还是“只是工具挂了”。希望帮到你。

本文还有配套的精品资源，点击获取