第八章 开发框架全景¶

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("将以下文本翻译成{language}: {text}")
llm = ChatOpenAI(model="gpt-5.5")
chain = prompt | llm | StrOutputParser()

result = chain.invoke({"language": "英文", "text": "你好世界"})

from langchain.agents import create_agent
from langchain.tools import tool

@tool
def get_order_status(order_id: str) -> str:
    """Return order status."""
    return f"{order_id}: shipped"

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[get_order_status],
    system_prompt="你是客服助理，需要时先调用工具再回答。",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "帮我查一下 ORD-001"}]}
)
print(result["messages"][-1].content)

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[lookup_order, refund_order],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "lookup_order": False,
                "refund_order": {
                    "allowed_decisions": ["approve", "edit", "reject"]
                },
            }
        )
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "customer-123"}}

paused = agent.invoke(
    {"messages": [{"role": "user", "content": "给 ORD-001 办理退款"}]},
    config=config,
    version="v2",
)

approved = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
    version="v2",
)

from langchain.chat_models import init_chat_model
from langchain.messages import ToolMessage
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.types import Command, interrupt

model = init_chat_model("openai:gpt-5.5", temperature=0)

@tool
def lookup_order(order_id: str) -> str:
    """Look up the order."""
    return f"{order_id}: delivered"

@tool
def refund_order(order_id: str, reason: str) -> str:
    """Submit refund request."""
    return f"{order_id}: refund requested for {reason}"

tools = [lookup_order, refund_order]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

class CustomerState(MessagesState):
    customer_id: str

def llm_node(state: CustomerState):
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def tool_or_review_node(state: CustomerState):
    results = []
    for call in state["messages"][-1].tool_calls:
        if call["name"] == "refund_order":
            decision = interrupt(
                {
                    "kind": "refund_review",
                    "tool": call["name"],
                    "args": call["args"],
                    "allowed_decisions": ["approve", "edit", "reject"],
                }
            )
            if decision["type"] == "reject":
                results.append(
                    ToolMessage(
                        content=f"人工拒绝退款：{decision['message']}",
                        tool_call_id=call["id"],
                    )
                )
                continue
            if decision["type"] == "edit":
                call = {**call, "args": decision["edited_action"]["args"]}

        tool = tools_by_name[call["name"]]
        output = tool.invoke(call["args"])
        results.append(ToolMessage(content=output, tool_call_id=call["id"]))
    return {"messages": results}

def should_continue(state: CustomerState):
    return "tool_or_review" if state["messages"][-1].tool_calls else END

builder = StateGraph(CustomerState)
builder.add_node("llm", llm_node)
builder.add_node("tool_or_review", tool_or_review_node)
builder.add_edge(START, "llm")
builder.add_conditional_edges("llm", should_continue, ["tool_or_review", END])
builder.add_edge("tool_or_review", "llm")

graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "customer-123"}}

paused = graph.invoke(
    {
        "messages": [{"role": "user", "content": "先查询 ORD-001，必要时退款"}],
        "customer_id": "customer-123",
    },
    config=config,
)

resumed = graph.invoke(
    Command(
        resume={
            "type": "edit",
            "edited_action": {
                "name": "refund_order",
                "args": {"order_id": "ORD-001", "reason": "duplicate charge"},
            },
        }
    ),
    config=config,
)

from langgraph.graph import START, END, StateGraph

research_builder = StateGraph(ResearchState)
research_builder.add_node("search", search_node)
research_builder.add_node("summarize", summarize_node)
research_builder.add_edge(START, "search")
research_builder.add_edge("search", "summarize")
research_builder.add_edge("summarize", END)

research_graph = research_builder.compile()
main_builder.add_node("research", research_graph)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "帮我处理退款"}]},
    config=config,
    stream_mode=["updates", "messages"],
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        if token.content:
            print(token.content, end="", flush=True)
    elif chunk["type"] == "updates" and "__interrupt__" in chunk["data"]:
        print("\n等待人工审核...")

from langgraph.types import RetryPolicy, TimeoutPolicy
from langgraph.errors import NodeError

def on_llm_failed(state: CustomerState, error: NodeError):
    # 仅在重试全部耗尽后触发;error.node / error.error 提供失败上下文
    return {"messages": [{"role": "assistant", "content": "系统繁忙，请稍后再试。"}]}

builder.add_node(
    "llm",
    llm_node,
    retry_policy=RetryPolicy(max_attempts=4, backoff_factor=2.0),  # 指数退避重试瞬时故障
    timeout=TimeoutPolicy(run_timeout=30, idle_timeout=5),         # 墙钟超时 + 无进展超时
    error_handler=on_llm_failed,                                   # 兜底:降级回复 / 转人工 / 补偿
)

from llama_index.core import VectorStoreIndex, Document

# 从文档创建向量索引

documents = [
    Document(text="人工智能的历史可以追溯到1950年代..."),
    Document(text="机器学习是人工智能的一个分支..."),
]

index = VectorStoreIndex.from_documents(documents)

# 创建查询引擎

query_engine = index.as_query_engine(
    similarity_top_k=5,
    response_mode="tree_summarize"
)

# 执行查询

response = query_engine.query("什么是机器学习？")
print(response)

from llama_index.core.postprocessor import (
    SimilarityPostprocessor,
    LongContextReorder
)

query_engine = index.as_query_engine(
    similarity_top_k=10,
    node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7),
        LongContextReorder()  # 重排序优化长上下文
    ]
)

from llama_index.core.agent.workflow import ReActAgent
from llama_index.core.tools import QueryEngineTool
from llama_index.core.workflow import Context
from llama_index.llms.openai import OpenAI

# 配置LLM
llm = OpenAI(model="gpt-5.5")

# 将查询引擎封装为工具
query_engine_tool = QueryEngineTool.from_defaults(
    query_engine=query_engine,
    name="knowledge_base",
    description="用于查询公司内部知识库的工具"
)

# 添加其他工具
tools = [
    query_engine_tool,
    search_web_tool,
    calculator_tool
]

# 创建智能体
agent = ReActAgent(
    tools=tools,
    llm=llm,
    verbose=True
)

# 使用
ctx = Context(agent)
response = await agent.run(user_msg="公司的休假政策是什么？", ctx=ctx)

from llama_index.core import VectorStoreIndex
from llama_index.core.tools import QueryEngineTool
from llama_index.core.agent.workflow import ReActAgent
from llama_index.core.workflow import Context
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-5.5")

# 为不同数据源创建索引
hr_index = VectorStoreIndex.from_documents(hr_docs)
finance_index = VectorStoreIndex.from_documents(finance_docs)
tech_index = VectorStoreIndex.from_documents(tech_docs)

# 创建多个查询引擎工具
tools = [
    QueryEngineTool.from_defaults(
        query_engine=hr_index.as_query_engine(),
        name="hr_knowledge",
        description="人力资源相关问题：休假、福利、考勤"
    ),
    QueryEngineTool.from_defaults(
        query_engine=finance_index.as_query_engine(),
        name="finance_knowledge",
        description="财务相关问题：报销、预算、账务"
    ),
    QueryEngineTool.from_defaults(
        query_engine=tech_index.as_query_engine(),
        name="tech_knowledge",
        description="技术文档：API、架构、部署"
    )
]

# 智能体会自动选择合适的知识库
agent = ReActAgent(tools=tools, llm=llm)

ctx = Context(agent)
response = await agent.run(user_msg="请对比一下HR和财务相关的最新政策", ctx=ctx)

from llama_index.core.query_engine import SubQuestionQueryEngine
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-5.5")

# 创建子问题查询引擎
sub_question_engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=[
        QueryEngineTool(
            query_engine=sales_engine,
            metadata=ToolMetadata(
                name="sales_data",
                description="销售数据查询"
            )
        ),
        QueryEngineTool(
            query_engine=marketing_engine,
            metadata=ToolMetadata(
                name="marketing_data",
                description="营销数据查询"
            )
        )
    ],
    llm=llm
)

# 复杂问题会被自动分解
response = sub_question_engine.query(
    "对比去年和今年的销售额，分析营销活动的效果"
)

# 内部会生成子问题：
# 1. 去年的销售额是多少？
# 2. 今年的销售额是多少？
# 3. 去年的营销活动有哪些？
# 4. 今年的营销活动有哪些？
# 然后综合回答

from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    Settings
)
from llama_index.core.agent.workflow import ReActAgent
from llama_index.core.tools import QueryEngineTool, FunctionTool
from llama_index.core.workflow import Context
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# 1. 配置全局 Settings
Settings.llm = OpenAI(model="gpt-5.5")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# 2. 加载文档
documents = SimpleDirectoryReader(
    input_dir="./documents",
    recursive=True,
    required_exts=[".pdf", ".docx", ".xlsx", ".txt"]
).load_data()

# 3. 创建索引
index = VectorStoreIndex.from_documents(documents)

# 4. 定义工具
def summarize_document(doc_name: str) -> str:
    """生成指定文档的摘要"""
    # 过滤特定文档
    doc_index = VectorStoreIndex.from_documents(
        [d for d in documents if doc_name in d.metadata.get("file_name", "")]
    )
    engine = doc_index.as_query_engine(response_mode="tree_summarize")
    return engine.query("请总结这个文档的主要内容").response

def compare_documents(doc1: str, doc2: str, aspect: str) -> str:
    """对比两个文档在某个方面的差异"""
    query = f"对比 {doc1} 和 {doc2} 在 {aspect} 方面的差异"
    return index.as_query_engine().query(query).response

# 包装为工具
summarize_tool = FunctionTool.from_defaults(fn=summarize_document)
compare_tool = FunctionTool.from_defaults(fn=compare_documents)

# 5. 创建智能体
tools = [
    QueryEngineTool.from_defaults(
        query_engine=index.as_query_engine(),
        name="document_search",
        description="搜索文档内容，回答关于文档的问题"
    ),
    summarize_tool,
    compare_tool
]

agent = ReActAgent(
    tools=tools,
    llm=Settings.llm,
    verbose=True
)

# 6. 使用
ctx = Context(agent)
response = await agent.run(
    user_msg="帮我总结一下合同文档，并对比与上一版的主要变化",
    ctx=ctx
)
print(response)

# AG2 示例：使用当前文档中的 ConversableAgent + AutoPattern 风格

import logging
from autogen import ConversableAgent, LLMConfig
from autogen.agentchat import run_group_chat
from autogen.agentchat.group.patterns import AutoPattern

logging.basicConfig(level=logging.INFO)

llm_config = LLMConfig.from_json(path="OAI_CONFIG_LIST")

planner = ConversableAgent(
    name="planner",
    system_message="把需求拆成可执行的开发任务。",
    description="负责任务拆解与推进。",
    llm_config=llm_config,
)
coder = ConversableAgent(
    name="coder",
    system_message="根据任务编写代码，并说明如何运行与测试。",
    description="负责实现。",
    llm_config=llm_config,
)
reviewer = ConversableAgent(
    name="reviewer",
    system_message="审查代码、指出问题，满足要求时输出 DONE。",
    description="负责质量把关。",
    is_termination_msg=lambda x: "DONE" in (x.get("content", "") or "").upper(),
    llm_config=llm_config,
)

pattern = AutoPattern(
    agents=[planner, coder, reviewer],
    initial_agent=planner,
    group_manager_args={"name": "group_manager", "llm_config": llm_config},
)

response = run_group_chat(
    pattern=pattern,
    messages="做一个简单的贪吃蛇游戏，并给出运行说明。",
    max_rounds=12,
)
response.process()
print(response.summary)

# CrewAI 示例（1.14.x 系列 API）：用"角色 + 任务流"构建市场调研团队

from crewai import Agent, Task, Crew

researcher = Agent(
    role="研究员",
    goal="收集信息并给出可验证依据",
    backstory="你是一名严谨的研究员，擅长检索并核验信息来源。",
    verbose=True
)
writer = Agent(
    role="写作者",
    goal="把研究结果组织成可读文章",
    backstory="你是一名结构化写作专家，擅长把零散材料组织成清晰文章。",
    verbose=True
)

research_task = Task(
    description="调研某主题的关键趋势",
    agent=researcher,
    expected_output="包含可验证来源的调研报告"
)
write_article_task = Task(
    description="基于调研产出文章",
    agent=writer,
    expected_output="结构清晰的最终文章",
    context=[research_task]
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_article_task],
    verbose=True
)

crew_output = crew.kickoff()
print(crew_output.raw)
print(crew_output.tasks_output)

from semantic_kernel.functions import kernel_function

class InventoryPlugin:
    @kernel_function(
        description="根据产品ID查询当前库存数量",
        name="GetStockCount"
    )
    def get_stock_count(self, product_id: str) -> int:
        # 这里调用旧系统的 API 或数据库

        print(f"Checking database for {product_id}...")
        return 42

    @kernel_function(
        description="当库存低于阈值时发送补货邮件",
        name="SendRestockEmail"
    )
    def send_restock_email(self, product_id: str, quantity: int) -> str:
        return f"Email sent for {quantity} units of {product_id}"

# 定义一个写邮件的 Prompt 模板

email_prompt = """
请为产品 {{$product_name}} 写一封补货申请邮件。
当前库存: {{$current_stock}}
建议补货: {{$amount}}
语气: 正式且紧急
"""

# Semantic Kernel 1.x 示例：初始化 Kernel、注册服务与插件，并启用自动函数调用

import os
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai import FunctionChoiceBehavior
from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings

async def main():
    # 1. 初始化 Kernel

    kernel = Kernel()

    # 2. 配置 AI 服务（具体方法名请按当前 release notes 核对）

    service_id = "default"
    kernel.add_service(
        OpenAIChatCompletion(
            service_id=service_id,
            ai_model_id="gpt-5.5",
            api_key=os.getenv("OPENAI_API_KEY"),
        )
    )

    # 3. 注册插件

    # 导入原生插件

    kernel.add_plugin(InventoryPlugin(), "Inventory")

    # 导入语义插件 (Prompt)

    kernel.add_function(
        plugin_name="Office",
        function_name="WriteEmail",
        prompt=email_prompt
    )

    # 4. 启用自动函数调用
    settings = PromptExecutionSettings()
    settings.function_choice_behavior = FunctionChoiceBehavior.Auto()

    # 用户的模糊指令
    goal = "查询某产品的库存，如果低于阈值，就生成一封补货申请邮件。"

    # 模型会在对话循环中自动选择 Inventory 与 Office 函数
    result = await kernel.invoke_prompt(goal, settings=settings)

    print(result)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

client = UnifiedAPIClient()
response = client.create(
    model="...",
    input="分析上传的数据，并在需要时调用工具",
    tools=[...],
    store=True,
)

next_response = client.create(
    model="...",
    previous_response_id=response.id,
    input="继续，顺便给我结论摘要",
)

from claude_agent_sdk import query, ClaudeAgentOptions

async for message in query(
    prompt="Review this repository for security issues",
    options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"]),
):
    print(message)

import time
import tracemalloc

class AgentBenchmark:
    def __init__(self, name: str):
        self.name = name
        self.metrics = {}

    def __enter__(self):
        tracemalloc.start()
        self.start_time = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.metrics["latency_ms"] = (time.perf_counter() - self.start_time) * 1000
        current, peak = tracemalloc.get_traced_memory()
        self.metrics["peak_memory_mb"] = peak / 10**6
        tracemalloc.stop()
        # 实际生产中应写入 Prometheus 或可观测背板
        print(
            f"[{self.name}] Latency: {self.metrics['latency_ms']:.2f}ms | "
            f"Peak RAM: {self.metrics['peak_memory_mb']:.2f}MB"
        )

# 使用示例：
# with AgentBenchmark("Agent_Tool_Call_Chain"):
#     response = agent_executor.invoke("分析财务数据并绘图")