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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_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_urlhttps://taotoken.net/api, api_keysk-你的TaoToken密钥 ) def call_llm(system: str, user: str) - str: resp client.chat.completions.create( modelclaude-sonnet-4-20250514, messages[ {role: system, content: system}, {role: user, content: user} ], temperature0.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, totalFalse): 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(hours2)).isoformat(), attendees: [{name: Alice, org: ExampleCorp}], is_external: True }] def fetch_gmail(): return [{subject: RFP: 周五前需要方案, from: clientexample.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_asciiFalse)} 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(checkpointerMemorySaver())这套图跑起来后你不需要手动传 goal它自己从上下文里找。这就是主动性和被动问答的本质区别。4. 验证请求与成功结果图搭好了怎么确认它真的在“主动工作”我一般分两步验证。第一步直接跑一次 invoke看 triggered 和 goal 是否被正确填充app build_graph() config {configurable: {thread_id: user-123}} out app.invoke({}, configconfig) 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第 {i1} 轮 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_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。如果你想先在网页里验证模型对话效果确认模型名和参数没问题可以用模型对话入口https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。接入过程中遇到 Key 或端点问题直接看接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。需要管理多个 Key 或查看用量去控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。最后说一个实用技巧主动性的触发规则不要一上来就写得太复杂。先用一两条硬规则比如“有外部会议就准备简报”跑通闭环确认 trace 完整、状态累积正常再逐步加规则或引入模型判断。规则越简单调试越快等骨架稳了再让它“聪明”起来。
