企业官网建设流程全解析

1. 从“会聊天”到“会自己找活干”：Photon AI 场景下的主动性 Agent

如果你用过 Photon AI 这类对话式产品，大概率会有一种感觉：它回答问题很利索，但你得一直盯着它、一直喂指令。你问一句它答一句，你不问它就安静待着。这种形态严格来说叫 Chatbot，不叫 Agent。真正的智能体（Agent）应该具备一个核心特征——主动性（Proactivity）：在你开口之前，它已经根据你的日历、邮件、待办，判断出“今天下午有个外部会议，我得先把参会人背景查一遍”。

这篇内容聚焦一件事：用 LangGraph 搭一个具备主动性的终极形态 Agent，并把它接到 Photon AI 这类应用场景里。我会给出可复制的状态图骨架、节点配置、TaoToken 统一 Key 通道的 settings.json 示例，以及一套端到端的验证动作。适合已经写过基础 LangChain 链、想往“自主执行 + 长时运行”方向走一步的开发者。读完之后，你应该能跑出一个会自己触发任务、自己规划、自己反思的最小闭环。

2. 前置准备：TaoToken 统一 Key 与 API 通道

在写图之前，先把模型调用这条链路理顺。LangGraph 本身只是编排层，真正干活的是节点里的大模型。如果你在多个工具（Claude Code、Cline、Continue、自己的 Python 脚本）之间来回切换，每个都配一遍 Key 会很烦。TaoToken 的思路是提供一个统一的 API 通道，你拿一个 Key，就能在多个 AI 工具里复用。

官网入口在这里：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。注册后在控制台生成 API Key，地址是 https://taotoken.net/api ，注意这个 API 地址不带 UTM 参数，直接填就行。

对于 Photon AI 这类应用场景，你通常需要两种接入方式：一种是在编辑器/CLI 工具里通过 settings.json 配置，另一种是在 Python 代码里通过 OpenAI 兼容接口调用。下面分别给出。

2.1 settings.json 配置示例（编辑器/CLI 工具）

很多 AI 编码工具支持自定义 OpenAI 兼容端点。以常见的 settings.json 结构为例，你可以这样写：

{ "ai.providers": { "taotoken": { "baseUrl": "https://taotoken.net/api", "apiKey": "sk-你的TaoToken密钥", "models": { "default": "claude-sonnet-4-20250514", "fast": "gpt-4o-mini" } } }, "ai.defaultProvider": "taotoken" }

这里的关键是 baseUrl 指向 TaoToken 的 API 地址，apiKey 填你在控制台生成的 Key。模型名按你实际可用的填，不要照抄。配置完成后，工具里的所有请求都会走这条统一通道。

2.2 Python 侧调用（LangGraph 节点内使用）

LangGraph 节点里调用模型，用 OpenAI 兼容的客户端即可：

from openai import OpenAI client = OpenAI( base_url="https://taotoken.net/api", api_key="sk-你的TaoToken密钥" ) def call_llm(system: str, user: str) -> str: resp = client.chat.completions.create( model="claude-sonnet-4-20250514", messages=[ {"role": "system", "content": system}, {"role": "user", "content": user} ], temperature=0.2 ) return resp.choices[0].message.content

这样你的 LangGraph 节点就不依赖某一家厂商的 SDK，换模型只改 model 字段。如果你更习惯用 LangChain 的封装，也可以把 base_url 和 api_key 传给 ChatOpenAI，效果一样。

3. 可复制的 LangGraph 状态图骨架

现在进入核心部分。一个具备主动性的 Agent，它的图结构不能只是“输入→LLM→输出”，而应该包含感知、触发、规划、执行、反思、学习这几个阶段。下面是我实测下来比较稳的一套骨架。

3.1 状态定义

先定义 AgentState，用 TypedDict 承载整个流程的上下文：

from typing import TypedDict, List, Optional, Literal from pydantic import BaseModel, Field class Subtask(BaseModel): id: str title: str objective: str status: Literal["pending", "running", "done", "failed"] = "pending" result: Optional[str] = None class AgentState(TypedDict, total=False): ctx: dict # 连接器拉来的上下文 triggered: bool # 是否触发主动任务 goal: str # 触发后的目标 subtasks: List[Subtask] traces: List[dict] reflection: str memory: dict loop_count: int

这里 triggered 和 goal 是主动性的关键：trigger 节点负责判断“要不要干活”，而不是等用户输入。

3.2 节点实现

感知节点（sense）负责从连接器拉数据。生产环境里这里接 Gmail、日历、Notion 的 API，演示阶段用桩函数：

from datetime import datetime, timedelta def fetch_calendar(): now = datetime.now() return [{ "title": "外部合作方会议", "start": (now + timedelta(hours=2)).isoformat(), "attendees": [{"name": "Alice", "org": "ExampleCorp"}], "is_external": True }] def fetch_gmail(): return [{"subject": "RFP: 周五前需要方案", "from": "client@example.com", "unread": True}] def sense(state: AgentState) -> AgentState: state["ctx"] = { "now": datetime.now().isoformat(), "calendar": fetch_calendar(), "gmail": fetch_gmail() } state.setdefault("traces", []).append({"node": "sense"}) return state

触发节点（trigger）是主动性的灵魂。它不等人问，而是根据上下文规则判断：

def trigger(state: AgentState) -> AgentState: ctx = state["ctx"] triggered, goal = False, "" external = [e for e in ctx["calendar"] if e.get("is_external")] if external: triggered = True goal = f"为会议准备简报：{external[0]['title']}，参会人 {external[0]['attendees']}" if not triggered: rfp = [g for g in ctx["gmail"] if g.get("unread") and "RFP" in g.get("subject", "")] if rfp: triggered = True goal = f"起草 RFP 响应大纲：{rfp[0]['subject']}" state["triggered"] = triggered state["goal"] = goal state.setdefault("traces", []).append({"node": "trigger", "triggered": triggered, "goal": goal}) return state

规划节点（plan）让模型把目标拆成子任务。这里用 JSON 输出约束：

import json def plan(state: AgentState) -> AgentState: if not state.get("triggered"): return state system = """你是自主 Agent 的规划器。返回严格 JSON： {"subtasks": [{"id": "...", "title": "...", "objective": "..."}]} 生成 3-6 个子任务，尽量可并行。""" user = f"目标：{state['goal']}\n上下文：{json.dumps(state['ctx'], ensure_ascii=False)}" raw = call_llm(system, user) data = json.loads(raw) state["subtasks"] = [Subtask(**t) for t in data["subtasks"]] state.setdefault("traces", []).append({"node": "plan", "count": len(state["subtasks"])}) return state

执行节点（dispatch）遍历子任务，逐个调用工具或模型。反思节点（reflect）检查失败项，决定是否重规划。学习节点（learn）把本轮经验写进 memory。

3.3 图组装与条件路由

from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver def route_after_trigger(state: AgentState) -> str: return "plan" if state.get("triggered") else "idle" def route_after_reflect(state: AgentState) -> str: if state.get("triggered") and state.get("subtasks") == []: return "plan" return "learn" def build_graph(): g = StateGraph(AgentState) g.add_node("sense", sense) g.add_node("trigger", trigger) g.add_node("plan", plan) g.add_node("dispatch", dispatch) g.add_node("reflect", reflect) g.add_node("learn", learn) g.add_node("idle", lambda s: s) g.set_entry_point("sense") g.add_edge("sense", "trigger") g.add_conditional_edges("trigger", route_after_trigger, {"plan": "plan", "idle": "idle"}) g.add_edge("plan", "dispatch") g.add_edge("dispatch", "reflect") g.add_conditional_edges("reflect", route_after_reflect, {"plan": "plan", "learn": "learn"}) g.add_edge("learn", "idle") g.add_edge("idle", END) return g.compile(checkpointer=MemorySaver())

这套图跑起来后，你不需要手动传 goal，它自己从上下文里找。这就是主动性和被动问答的本质区别。

4. 验证请求与成功结果

图搭好了，怎么确认它真的在“主动工作”？我一般分两步验证。

第一步，直接跑一次 invoke，看 triggered 和 goal 是否被正确填充：

app = build_graph() config = {"configurable": {"thread_id": "user-123"}} out = app.invoke({}, config=config) print("Triggered:", out.get("triggered")) print("Goal:", out.get("goal")) print("Subtasks:", len(out.get("subtasks", []))) print("Trace nodes:", [t["node"] for t in out.get("traces", [])])

预期输出类似：

Triggered: True Goal: 为会议准备简报：外部合作方会议，参会人 [{'name': 'Alice', 'org': 'ExampleCorp'}] Subtasks: 4 Trace nodes: ['sense', 'trigger', 'plan', 'dispatch', 'reflect', 'learn']

如果 Triggered 是 True，说明它没等你输入就自己找到了任务。Trace 里节点顺序完整，说明整条链路走通了。

第二步，验证长时运行。把 invoke 包在一个循环里，模拟外部调度器每 N 分钟触发一次：

import time for i in range(3): out = app.invoke({}, config={"configurable": {"thread_id": "user-123"}}) print(f"第 {i+1} 轮 loop_count:", out.get("loop_count")) time.sleep(2)

因为用了 MemorySaver，同一个 thread_id 下的状态会累积，loop_count 会递增。生产环境把 MemorySaver 换成 Redis 或 Postgres 的 checkpointer，就能实现真正的 7x24 持久运行。

5. 本篇常见错排查

跑这套东西，我踩过的坑集中在几个地方，列出来帮你省时间。

报错一：json.decoder.JSONDecodeError。模型返回的 JSON 里带了 markdown 代码块标记，比如json ...。解决办法是在解析前先剥掉围栏，或者用正则提取第一个{到最后一个}之间的内容。更稳的做法是用结构化输出（JSON schema），让模型直接吐合法 JSON。

报错二：triggered一直是 False。检查你的连接器桩函数返回的数据结构，字段名要和 trigger 节点里读的一致。比如你写的是is_external，但桩函数里写成了external，条件永远不成立。建议在 sense 节点后打印一次 ctx 确认。

报错三：条件路由报KeyError。add_conditional_edges的映射字典必须覆盖路由函数所有可能的返回值。如果你 route_after_trigger 可能返回 "plan" 和 "idle"，映射里两个都要有，少一个就报错。

报错四：MemorySaver 状态不累积。确认每次 invoke 传的是同一个 thread_id。不同 thread_id 是隔离的，状态不会串。另外 MemorySaver 只在进程内有效，重启就丢，别拿它当生产存储。

报错五：模型调用超时或 401。先确认 base_url 和 api_key 配对正确。TaoToken 的 API 地址是 https://taotoken.net/api ，不要多加路径。Key 从控制台复制时注意别带空格。如果还是 401，去 https://taotoken.net/api-keys 重新生成一个。

6. 把这条链路接进你的工作流

到这里，你已经有了一个能自己感知、自己触发、自己规划、自己反思的最小 Agent。接下来是把它用起来。

如果你主要是在编辑器或 CLI 里做长期编码、跑 Agent 任务，建议把 TaoToken 的 Key 配到 Coding Plan 里，这样多个工具共享一条通道，不用反复切换：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。

如果你想先在网页里验证模型对话效果，确认模型名和参数没问题，可以用模型对话入口：https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。

接入过程中遇到 Key 或端点问题，直接看接入文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。需要管理多个 Key 或查看用量，去控制台：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite 。

最后说一个实用技巧：主动性的触发规则不要一上来就写得太复杂。先用一两条硬规则（比如“有外部会议就准备简报”）跑通闭环，确认 trace 完整、状态累积正常，再逐步加规则或引入模型判断。规则越简单，调试越快，等骨架稳了再让它“聪明”起来。