1. LangGraph实战

# 分解问题链
decomposition_chain = (
    {"question": RunnablePassthrough()}
    | decomposition_prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
    | (lambda x: x.strip().split("\n"))
)

# 子问题答案生成链
sub_question_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | sub_question_prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

# 组装链
chain = (
    {
        "question": RunnablePassthrough(),
        "context": decomposition_chain | {
            "questions": RunnablePassthrough(),
            "answers": sub_question_chain.map()
        } | RunnableLambda(format_qa_pairs)
    } | prompt | llm_output_str
)

from typing import List, Dict
from pydantic import BaseModel

class ChatState(BaseModel):
    messages: List[Dict[str, str]] = []
    current_input: str = ""
    tools_output: Dict[str, str] = {}
    final_response: str = ""

async def process_input(state: ChatState) -> ChatState:
    # 处理用户输入
    messages = state.messages + [{"role": "user", "content": state.current_input}]
    return ChatState(
        messages=messages,
        current_input=state.current_input,
        tools_output=state.tools_output
    )

async def generate_response(state: ChatState) -> ChatState:
    # 使用LLM生成回复
    response = await llm.ainvoke(state.messages)
    messages = state.messages + [{"role": "assistant", "content": response}]
    return ChatState(
        messages=messages,
        current_input=state.current_input,
        tools_output=state.tools_output,
        final_response=response
    )

from langgraph.graph import StateGraph, END

# 创建图结构
workflow = StateGraph(ChatState)

# 添加节点
workflow.add_node("process_input", process_input)
workflow.add_node("generate_response", generate_response)

# 定义边和路由逻辑
workflow.add_edge("process_input", "generate_response")
workflow.add_edge("generate_response", END)

from typing import List, Dict, Tuple
from pydantic import BaseModel
from langgraph.graph import StateGraph, END
from langchain_core.language_models import ChatOpenAI

# 1. 定义状态
class ChatState(BaseModel):
    messages: List[Dict[str, str]] = []
    current_input: str = ""
    should_continue: bool = True

# 2. 定义节点函数
async def process_user_input(state: ChatState) -> ChatState:
    """处理用户输入"""
    messages = state.messages + [{"role": "user", "content": state.current_input}]
    return ChatState(
        messages=messages,
        current_input=state.current_input,
        should_continue=True
    )

async def generate_ai_response(state: ChatState) -> ChatState:
    """生成AI回复"""
    llm = ChatOpenAI(temperature=0.7)
    response = await llm.ainvoke(state.messages)
    messages = state.messages + [{"role": "assistant", "content": response}]
    return ChatState(
        messages=messages,
        current_input=state.current_input,
        should_continue=True
    )

def should_continue(state: ChatState) -> str:
    """决定是否继续对话"""
    if "goodbye" in state.current_input.lower():
        return "end"
    return "continue"

# 3. 构建图
workflow = StateGraph(ChatState)

# 添加节点
workflow.add_node("process_input", process_user_input)
workflow.add_node("generate_response", generate_ai_response)

# 添加边
workflow.add_edge("process_input", "generate_response")
workflow.add_conditional_edges(
    "generate_response",
    should_continue,
    {
        "continue": "process_input",
        "end": END
    }
)

# 4. 编译图
app = workflow.compile()

# 5. 运行对话
async def chat():
    state = ChatState()
    while True:
        user_input = input("You: ")
        state.current_input = user_input
        state = await app.ainvoke(state)
        print("Bot:", state.messages[-1]["content"])
        if not state.should_continue:
            break

# 运行聊天
import asyncio
asyncio.run(chat())

from typing import List, Dict, Literal
from pydantic import BaseModel
from langgraph.graph import StateGraph, END

class AgentState(BaseModel):
    messages: List[Dict[str, str]] = []
    current_input: str = ""
    tools_output: Dict[str, str] = {}
    status: str = "RUNNING"
    error_count: int = 0

def route_by_status(state: AgentState) -> Literal["process", "retry", "error", "end"]:
    """复杂的路由逻辑"""
    if state.status == "SUCCESS":
        return "end"
    elif state.status == "ERROR":
        if state.error_count >= 3:
            return "error"
        return "retry"
    elif state.status == "NEED_TOOL":
        return "process"
    return "process"

# 构建图结构
workflow = StateGraph(AgentState)

# 添加条件边
workflow.add_conditional_edges(
    "check_status",
    route_by_status,
    {
        "process": "execute_tool",
        "retry": "retry_handler",
        "error": "error_handler",
        "end": END
    }
)

async def parallel_tools_execution(state: AgentState) -> AgentState:
    """并行执行多个工具"""
    tools = identify_required_tools(state.current_input)
    
    async def execute_tool(tool):
        result = await tool.ainvoke(state.current_input)
        return {tool.name: result}
    
    # 并行执行所有工具
    results = await asyncio.gather(*[execute_tool(tool) for tool in tools])
    
    # 合并结果
    tools_output = {}
    for result in results:
        tools_output.update(result)
    
    return AgentState(
        messages=state.messages,
        current_input=state.current_input,
        tools_output=tools_output,
        status="SUCCESS"
    )

# 创建图结构
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("think", think)
workflow.add_node("execute_tool", execute_tool)
workflow.add_node("generate_response", generate_response)

# 定义路由函数
def route_next_step(state: AgentState) -> str:
    if state.status == "ERROR":
        return "error"
    elif state.status == "NEED_TOOL":
        return "execute_tool"
    elif state.status == "GENERATE_RESPONSE":
        return "generate_response"
    elif state.status == "SUCCESS":
        return "end"
    return "think"

# 添加条件边
workflow.add_conditional_edges(
    "think",
    route_next_step,
    {
        "execute_tool": "execute_tool",
        "generate_response": "generate_response",
        "error": "error_handler",
        "end": END
    }
)

workflow.add_conditional_edges(
    "execute_tool",
    route_next_step,
    {
        "generate_response": "generate_response",
        "error": "error_handler",
        "end": END
    }
)

workflow.add_edge("generate_response", END)

# 编译图
app = workflow.compile()

async def run_agent(user_input: str):
    state = AgentState(current_input=user_input)
    final_state = await app.ainvoke(state)
    return final_state.final_answer

# 使用示例
async def main():
    questions = [
        "计算23乘以45等于多少？",
        "谁发明了相对论？",
        "计算圆周率乘以10等于多少？"
    ]
    
    for question in questions:
        print(f"\n问题: {question}")
        answer = await run_agent(question)
        print(f"回答: {answer}")

# 运行示例
import asyncio
asyncio.run(main())

from langgraph.persist import GraphStatePersist

class DBStatePersist(GraphStatePersist):
    async def persist(self, state: AgentState):
        # 实现状态持久化逻辑
        pass
    
    async def load(self) -> AgentState:
        # 实现状态加载逻辑
        pass

# 使用持久化
workflow = StateGraph(AgentState, persist_handler=DBStatePersist())

async def stream_handler(state: AgentState):
    """处理流式输出"""
    yield f"思考中: {state.thought}\n"
    if state.selected_tool:
        yield f"使用工具: {state.selected_tool}\n"
    if state.tool_output:
        yield f"工具输出: {state.tool_output}\n"
    yield f"最终答案: {state.final_answer}\n"

# 在图中启用流式输出
workflow.set_stream_handler(stream_handler)

async def error_handler(state: AgentState) -> AgentState:
    """处理错误情况"""
    if state.error_count < 3:
        return AgentState(
            **state.dict(),
            error_count=state.error_count + 1,
            status="RETRY"
        )
    return AgentState(
        **state.dict(),
        final_answer="抱歉，我无法完成这个任务。",
        status="ERROR"
    )

# 添加错误处理节点
workflow.add_node("error_handler", error_handler)