AI Agents入门教程之Workflow vs Agent

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 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 = "\n\n---\n\n".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"])