AI Agent 编程实战：如何用动态执行图优化任务流

AI Agent 编程实战：如何用动态执行图优化任务流

为什么传统的 Agent 工作流不够用？

搭过几个 Agent 后，我越来越觉得“写死流程”这条路走不通。比如一个研发助手，逻辑看起来很清爽：

• 接收指令
• 搜索知识库
• 生成代码草稿
• 运行测试
• 汇报结果

五步走完，干净利落。但真实场景哪有这么乖？测试挂了要不要重试？搜不到资料是不是该换关键词？LLM 自己说“这题我不会”，要不要切到另一个专家模型？——这些判断没法提前写进 if 里。硬塞进去的结果就是：函数越写越长，分支嵌套三层起，改一行得翻十页日志。

动态执行图的核心思想

我的办法是把整个流程看成一张活的图：节点是动作（搜索、调用 API、写代码），边是“下一步去哪”。关键不是画好图就完事，而是让图在跑的时候自己决定怎么走。

它靠三样东西撑起来：

• 节点做完后不直接跳固定下一个，而是问一句：“现在状态怎样？该去哪？”答案可能是 A，也可能是 B，甚至原地打转。
• 所有节点共享一个 state 字典。前一步存的 search_result，后一步直接读，不用参数传参传到手抖。
• 循环不是靠 while True 硬套，而是让边指回前面的节点。比如“测试失败 → 回到生成代码”这种链路，图里画出来就是一条带箭头的线，清晰又自然。

如何落地？一个简单的实现框架

我一般照着 LangGraph 的路子来搭。下面这段伪代码，是我第一次跑通动态路由时抄在小本本上的核心骨架：

from typing import TypedDict
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    task: str
    result: str
    success: bool

# 定义节点函数
def analyze_task(state: AgentState) -> AgentState:
    # 模拟 LLM 分析
    state['result'] = "Task analyzed"
    state['success'] = True
    return state

def execute_task(state: AgentState) -> AgentState:
    # 模拟执行
    if state['success']:
        state['result'] = "Execution completed successfully"
    return state

# 定义条件路由函数
def check_success(state: AgentState) -> str:
    return "execute" if state['success'] else "retry"

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("analyze", analyze_task)
workflow.add_node("execute", execute_task)
workflow.add_node("retry", lambda s: s) # 简化的重试节点

# 设置入口和条件边
workflow.set_entry_point("analyze")
workflow.add_conditional_edges(
    "analyze",
    check_success,
    {
        "execute": "execute",
        "retry": "retry"
    }
)
workflow.add_edge("execute", END)
workflow.add_edge("retry", "analyze") # 重试后回到分析步骤

app = workflow.compile()

重点就在这句 add_conditional_edges。它让“走哪条路”这件事，从写代码时的静态决定，变成了运行时的一次函数调用。图活了。

动态图的实战收益

上线跑了一阵子，最明显的感受有三个：

• 测试失败不用人盯屏了。最多重试三次，失败了才弹告警。之前得手动重跑整条链，现在它自己兜底。
• 出问题时查得快。点开某次执行的历史，每个节点输入输出全在，一眼看出是搜索没返回内容，还是代码生成环节漏判了边界条件。
• 加新功能变简单了。上周要加个“自动提 PR”环节，我就加了个 create_pr 节点，连上 execute 后面，再补个条件判断是否需要提——没动其他地方。

下一步建议

如果你正用 LangChain 或类似工具写 Agent，别一上来就搞多智能体。先拿一个 DAG 试试水：两个节点，一条条件边，跑通再说。等你看着 state 在图里流来流去，心里有数了，再往上加循环、加并行、加人工干预钩子。

相关阅读：深入了解 LangGraph 的状态管理模型，探索更复杂的多智能体协作模式。