AI Agents入门教程之Workflow vs Agent

from langchain_openai import ChatOpenAI
if __name__ == "__main__":    llm = ChatOpenAI(        model_name = "qwen2.5:7b",        openai_api_key = "test",        openai_api_base = "https://localhost:11444/v1",        temperature = 0    )    answer = llm.invoke("who are you?")    print(answer.content)

结果如下：

05

智能体系统的基础架构始于一个通过检索、工具调用、记忆存储等能力增强的大语言模型（LLM）。

当前主流模型已能主动运用这些能力——自主生成搜索查询、筛选适用工具、判断信息留存策略。

在实现这些增强功能时，我们建议重点关注两个实现维度：

• 根据具体用例定制增强能力
• 为LLM提供简洁、文档完善的交互接口

虽然实现方式多样，其中一种典型方案是通过最新发布的模型上下文协议（Model Context Protocol）——该协议支持开发者通过轻量级客户端实现与日益丰富的第三方工具生态集成。

在本文后续讨论中，我们默认每次LLM调用均已接入上述增强能力。

# Schema for structured outputfrom pydantic import BaseModel, Field
class SearchQuery(BaseModel):    search_query: str = Field(None, description="Query that is optimized web search.")    justification: str = Field(        None, description="Why this query is relevant to the user's request."    )

# Augment the LLM with schema for structured outputstructured_llm = llm.with_structured_output(SearchQuery)
# Invoke the augmented LLMoutput = structured_llm.invoke("How does Calcium CT score relate to high cholesterol?")print(output)

# Define a tooldef multiply(a: int, b: int) -> int:    return a * b
# Augment the LLM with toolsllm_with_tools = llm.bind_tools([multiply])# Invoke the LLM with input that triggers the tool callmsg = llm_with_tools.invoke("What is 2 times 3?")# Get the tool callprint(msg.tool_calls)

结果如下：

06

在提示链(prompt chaining)中，每个LLM调用都会处理前一个步骤的输出结果。

from typing_extensions import TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langchain_openai import ChatOpenAI


llm = ChatOpenAI(        model_name = "qwen2.5:7b",        openai_api_key = "test",        openai_api_base = "https://localhost:11444/v1",        temperature = 0    )

# Graph stateclass State(TypedDict):    topic: str    joke: str    improved_joke: str    final_joke: str

# Nodesdef generate_joke(state: State):    """First LLM call to generate initial joke"""
    msg = llm.invoke(f"Write a short joke about {state['topic']}")    return {"joke": msg.content}

def check_punchline(state: State):    """Gate function to check if the joke has a punchline"""
    # Simple check - does the joke contain "?" or "!"    if "?" in state["joke"] or "!" in state["joke"]:        return "Fail"    return "Pass"

def improve_joke(state: State):    """Second LLM call to improve the joke"""
    msg = llm.invoke(f"Make this joke funnier by adding wordplay: {state['joke']}")    return {"improved_joke": msg.content}

def polish_joke(state: State):    """Third LLM call for final polish"""
    msg = llm.invoke(f"Add a surprising twist to this joke: {state['improved_joke']}")    return {"final_joke": msg.content}

# Build workflowworkflow = StateGraph(State)
# Add nodesworkflow.add_node("generate_joke", generate_joke)workflow.add_node("improve_joke", improve_joke)workflow.add_node("polish_joke", polish_joke)
# Add edges to connect nodesworkflow.add_edge(START, "generate_joke")workflow.add_conditional_edges(    "generate_joke", check_punchline, {"Fail": "improve_joke", "Pass": END})workflow.add_edge("improve_joke", "polish_joke")workflow.add_edge("polish_joke", END)
# Compilechain = workflow.compile()
# Save workflowimg = chain.get_graph().draw_png()with open("graph.jpg","wb") as f:    f.write(img)
# Invokestate = chain.invoke({"topic": "cats"})print("Initial joke:")print(state["joke"])print("n--- --- ---n")if "improved_joke" in state:    print("Improved joke:")    print(state["improved_joke"])    print("n--- --- ---n")
    print("Final joke:")    print(state["final_joke"])else:    print("Joke failed quality gate - no punchline detected!")

from typing import TypedDictfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph,START,END# Graph stateclass State(TypedDict):    topic: str    joke: str    story: str    poem: str    combined_output: str# llmllm = ChatOpenAI(        model_name = "qwen2.5:7b",        openai_api_key = "test",        openai_api_base = "https://localhost:11444/v1",        temperature = 0    )# Nodesdef call_llm_1(state: State):    """First LLM call to generate initial joke"""    msg = llm.invoke(f"Write a joke about {state['topic']}")    return {"joke": msg.content}def call_llm_2(state: State):    """Second LLM call to generate story"""    msg = llm.invoke(f"Write a story about {state['topic']}")    return {"story": msg.content}def call_llm_3(state: State):    """Third LLM call to generate poem"""    msg = llm.invoke(f"Write a poem about {state['topic']}")    return {"poem": msg.content}def aggregator(state: State):    """Combine the joke and story into a single output"""    combined = f"Here's a story, joke, and poem about {state['topic']}!nn"    combined += f"STORY:n{state['story']}nn"    combined += f"JOKE:n{state['joke']}nn"    combined += f"POEM:n{state['poem']}"    return {"combined_output": combined}# Build workflowparallel_builder = StateGraph(State)# Add nodesparallel_builder.add_node("call_llm_1", call_llm_1)parallel_builder.add_node("call_llm_2", call_llm_2)parallel_builder.add_node("call_llm_3", call_llm_3)parallel_builder.add_node("aggregator", aggregator)# Add edges to connect nodesparallel_builder.add_edge(START, "call_llm_1")parallel_builder.add_edge(START, "call_llm_2")parallel_builder.add_edge(START, "call_llm_3")parallel_builder.add_edge("call_llm_1", "aggregator")parallel_builder.add_edge("call_llm_2", "aggregator")parallel_builder.add_edge("call_llm_3", "aggregator")parallel_builder.add_edge("aggregator", END)parallel_workflow = parallel_builder.compile()# Save workflowimg = parallel_workflow.get_graph().draw_png()with open("parallel_graph.jpg","wb") as f:    f.write(img)# Invokestate = parallel_workflow.invoke({"topic": "cats"})print(state["combined_output"])

运行后工作流图示如下：

调用后输出如下：

from typing import TypedDictfrom typing_extensions import Literalfrom pydantic import BaseModel,Fieldfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph,START,ENDfrom langchain_core.messages import HumanMessage, SystemMessage

# Schema for structured output to use as routing logicclass Route(BaseModel):    step: Literal["poem", "story", "joke"] = Field(        None, description="The next step in the routing process"    )

# Augment the LLM with schema for structured outputllm = ChatOpenAI(        model_name = "qwen2.5:7b",        openai_api_key = "test",        openai_api_base = "https://localhost:11444/v1",        temperature = 0    )router = llm.with_structured_output(Route)

# Stateclass State(TypedDict):    input: str    decision: str    output: str

# Nodesdef llm_call_1(state: State):    """Write a story"""
    result = llm.invoke(state["input"])    return {"output": result.content}

def llm_call_2(state: State):    """Write a joke"""
    result = llm.invoke(state["input"])    return {"output": result.content}

def llm_call_3(state: State):    """Write a poem"""
    result = llm.invoke(state["input"])    return {"output": result.content}

def llm_call_router(state: State):    """Route the input to the appropriate node"""
    # Run the augmented LLM with structured output to serve as routing logic    decision = router.invoke(        [            SystemMessage(                content="Route the input to story, joke, or poem based on the user's request."            ),            HumanMessage(content=state["input"]),        ]    )
    return {"decision": decision.get("step")}

# Conditional edge function to route to the appropriate nodedef route_decision(state: State):    # Return the node name you want to visit next    if state["decision"] == "story":        return "llm_call_1"    elif state["decision"] == "joke":        return "llm_call_2"    elif state["decision"] == "poem":        return "llm_call_3"

# Build workflowrouter_builder = StateGraph(State)
# Add nodesrouter_builder.add_node("llm_call_1", llm_call_1)router_builder.add_node("llm_call_2", llm_call_2)router_builder.add_node("llm_call_3", llm_call_3)router_builder.add_node("llm_call_router", llm_call_router)
# Add edges to connect nodesrouter_builder.add_edge(START, "llm_call_router")router_builder.add_conditional_edges(    "llm_call_router",    route_decision,    {  # Name returned by route_decision : Name of next node to visit        "llm_call_1": "llm_call_1",        "llm_call_2": "llm_call_2",        "llm_call_3": "llm_call_3",    },)router_builder.add_edge("llm_call_1", END)router_builder.add_edge("llm_call_2", END)router_builder.add_edge("llm_call_3", END)
# Compile workflowrouter_workflow = router_builder.compile()
# Save the workflowimg = router_workflow.get_graph().draw_png()with open("router_graph.jpg","wb") as f:    f.write(img)
# Invokestate = router_workflow.invoke({"input": "Write me a joke about cats"})print(state["output"])

from typing import Annotated, Listimport operator

# Schema for structured output to use in planningclass Section(BaseModel):    name: str = Field(        description="Name for this section of the report.",    )    description: str = Field(        description="Brief overview of the main topics and concepts to be covered in this section.",    )

class Sections(BaseModel):    sections: List[Section] = Field(        description="Sections of the report.",    )

# Augment the LLM with schema for structured outputfrom langchain_anthropic import ChatAnthropicllm = ChatAnthropic(model="claude-3-5-sonnet-latest")planner = llm.with_structured_output(Sections)

from langgraph.constants import Send

# Graph stateclass State(TypedDict):    topic: str  # Report topic    sections: list[Section]  # List of report sections    completed_sections: Annotated[        list, operator.add    ]  # All workers write to this key in parallel    final_report: str  # Final report

# Worker stateclass WorkerState(TypedDict):    section: Section    completed_sections: Annotated[list, operator.add]

# Nodesdef orchestrator(state: State):    """Orchestrator that generates a plan for the report"""
    # Generate queries    report_sections = planner.invoke(        [            SystemMessage(content="Generate a plan for the report."),            HumanMessage(content=f"Here is the report topic: {state['topic']}"),        ]    )
    return {"sections": report_sections.sections}

def llm_call(state: WorkerState):    """Worker writes a section of the report"""
    # Generate section    section = llm.invoke(        [            SystemMessage(                content="Write a report section following the provided name and description. Include no preamble for each section. Use markdown formatting."            ),            HumanMessage(                content=f"Here is the section name: {state['section'].name} and description: {state['section'].description}"            ),        ]    )
    # Write the updated section to completed sections    return {"completed_sections": [section.content]}

def synthesizer(state: State):    """Synthesize full report from sections"""
    # List of completed sections    completed_sections = state["completed_sections"]
    # Format completed section to str to use as context for final sections    completed_report_sections = "nn---nn".join(completed_sections)
    return {"final_report": completed_report_sections}

# Conditional edge function to create llm_call workers that each write a section of the reportdef assign_workers(state: State):    """Assign a worker to each section in the plan"""
    # Kick off section writing in parallel via Send() API    return [Send("llm_call", {"section": s}) for s in state["sections"]]

# Build workfloworchestrator_worker_builder = StateGraph(State)
# Add the nodesorchestrator_worker_builder.add_node("orchestrator", orchestrator)orchestrator_worker_builder.add_node("llm_call", llm_call)orchestrator_worker_builder.add_node("synthesizer", synthesizer)
# Add edges to connect nodesorchestrator_worker_builder.add_edge(START, "orchestrator")orchestrator_worker_builder.add_conditional_edges(    "orchestrator", assign_workers, ["llm_call"])orchestrator_worker_builder.add_edge("llm_call", "synthesizer")orchestrator_worker_builder.add_edge("synthesizer", END)
# Compile the workfloworchestrator_worker = orchestrator_worker_builder.compile()
# Show the workflowdisplay(Image(orchestrator_worker.get_graph().draw_mermaid_png()))
# Invokestate = orchestrator_worker.invoke({"topic": "Create a report on LLM scaling laws"})
from IPython.display import MarkdownMarkdown(state["final_report"])

添加个人微信，进专属粉丝群！