同一个任务,
三种代码形状

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from pydantic import BaseModel
 
class Report(BaseModel):
    title: str
    summary: str
    findings: list[str]
    sources: list[str]
 
class State(TypedDict):
    query: str
    search_results: list
    summary: str
    report: Report | None
 
llm = ChatOpenAI(model="gpt-4o")
 
def search_node(s: State) -> State:
    s["search_results"] = web_search(s["query"])
    return s
 
def summarize_node(s: State) -> State:
    content = "\n".join(r["snippet"] for r in s["search_results"])
    s["summary"] = llm.invoke(f"总结: {content}").content
    return s
 
def structure_node(s: State) -> State:
    s["report"] = llm.with_structured_output(Report).invoke(
        f"转成结构化报告: {s['summary']}"
    )
    return s
 
graph = StateGraph(State)
graph.add_node("search", search_node)
graph.add_node("summarize", summarize_node)
graph.add_node("structure", structure_node)
graph.set_entry_point("search")
graph.add_edge("search", "summarize")
graph.add_edge("summarize", "structure")
graph.add_edge("structure", END)
 
app = graph.compile()
result = app.invoke({
    "query": "harness engineering trends 2026",
    "search_results": [], "summary": "", "report": None,
})
print(result["report"].title)

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
from pydantic import BaseModel
 
class Report(BaseModel):
    title: str
    summary: str
    findings: list[str]
    sources: list[str]
 
researcher = Agent(
    role="Senior Research Analyst",
    goal="挖掘 {topic} 领域最新、最可靠的信息",
    backstory="你在技术分析领域有 10 年经验。",
    tools=[SerperDevTool()],
    verbose=True,
)
 
analyst = Agent(
    role="Report Writer",
    goal="把研究发现整理成结构化报告",
    backstory="你写过 Stratechery 风格的深度分析。",
)
 
research_task = Task(
    description="调研 {topic}，至少找 5 个权威来源",
    agent=researcher,
    expected_output="研究发现列表，每个带来源链接",
)
 
report_task = Task(
    description="基于研究，输出结构化 Report",
    agent=analyst,
    context=[research_task],
    output_pydantic=Report,
    expected_output="完整的 Report 对象",
)
 
crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, report_task],
    process=Process.sequential,
)
 
result = crew.kickoff(inputs={"topic": "harness engineering"})
report: Report = result.pydantic
print(report.title)

from smolagents import CodeAgent, HfApiModel, tool
from pydantic import BaseModel
 
class Report(BaseModel):
    title: str
    summary: str
    findings: list[str]
    sources: list[str]
 
@tool
def web_search(query: str) -> list[dict]:
    """
    搜索网络，返回结果列表。
 
    Args:
        query: 搜索关键词
    """
    return search_api(query)
 
@tool
def final_report(
    title: str, summary: str,
    findings: list[str], sources: list[str],
) -> dict:
    """
    产出结构化 Report。
 
    Args:
        title: 报告标题
        summary: 摘要
        findings: 关键发现列表
        sources: 来源 URL 列表
    """
    return Report(
        title=title, summary=summary,
        findings=findings, sources=sources,
    ).model_dump()
 
agent = CodeAgent(
    tools=[web_search, final_report],
    model=HfApiModel("Qwen/Qwen2.5-72B-Instruct"),
    additional_authorized_imports=["json"],
)
 
# agent 自己生成 Python 代码调工具、做推理
result = agent.run(
    "调研 harness engineering 2026 最新趋势，"
    "产出含 title、summary、findings(5个)、sources 的 Report。"
)
print(result)