掌握这些知识点，就能轻松开发Agent！每个知识点都配有Agent场景的实战代码。

---

第一部分：必须掌握的基础（⭐⭐⭐⭐⭐）

1. 类型注解（TypedDict）- Agent的状态定义

为什么Agent开发需要：LangGraph的状态必须用类型注解定义

基础语法

from typing import TypedDict

# 定义Agent的状态结构
class AgentState(TypedDict):
    name: str           # 字符串类型
    age: int            # 整数类型
    messages: list      # 列表类型
    is_active: bool     # 布尔类型
    data: dict          # 字典类型

Agent实战场景

from typing import TypedDict, Annotated
from operator import add

# 聊天机器人的状态
class ChatBotState(TypedDict):
    messages: Annotated[list, add]  # 聊天历史，自动追加
    user_name: str                   # 用户名
    intent: str                      # 用户意图
    context: dict                    # 上下文信息

# 使用状态
def chat_node(state: ChatBotState) -> ChatBotState:
    # 可以智能提示state有哪些字段
    user_name = state["user_name"]  # IDE会提示
    messages = state["messages"]
    
    return {
        "messages": ["AI的回复"],
        "intent": "问候"
    }

重点记忆：

• TypedDict：定义状态结构
• Annotated[list, add]：列表自动追加，不覆盖
• 函数参数和返回值都要类型注解：-> AgentState

---

2. 字典操作 - 状态管理的核心

为什么Agent开发需要：Agent的状态就是一个大字典

必会操作

# 创建字典
state = {
    "user_id": "123",
    "messages": [],
    "count": 0
}

# 读取值
user_id = state["user_id"]
user_id = state.get("user_id", "默认值")  # 推荐！不会报错

# 更新值
state["count"] = 1
state.update({"count": 2, "new_key": "value"})

# 检查key是否存在
if "user_id" in state:
    print("存在")

# 合并字典（Agent开发常用）
state1 = {"a": 1, "b": 2}
state2 = {"b": 3, "c": 4}
merged = {**state1, **state2}  # {"a": 1, "b": 3, "c": 4}

Agent实战：状态更新

def process_message(state: AgentState) -> AgentState:
    # 只返回要修改的字段
    return {
        "messages": state["messages"] + ["新消息"],
        "count": state.get("count", 0) + 1
    }
    
# 完整状态合并（LangGraph自动处理）
# 旧状态：{"messages": ["hi"], "count": 0, "user": "张三"}
# 返回：{"messages": ["hi", "新消息"], "count": 1}
# 合并后：{"messages": ["hi", "新消息"], "count": 1, "user": "张三"}

重点记忆：

• 用state.get(key, default)避免KeyError
• 用{**dict1, **dict2}合并字典
• 返回部分状态，LangGraph自动合并

---

3. 列表操作 - 消息历史管理

为什么Agent开发需要：对话历史、步骤记录都用列表

必会操作

messages = []

# 添加元素
messages.append("新消息")
messages.extend(["消息1", "消息2"])
messages = messages + ["消息3"]

# 访问元素
first = messages[0]        # 第一个
last = messages[-1]        # 最后一个
recent = messages[-3:]     # 最后3个

# 遍历
for msg in messages:
    print(msg)

# 过滤
long_msgs = [m for m in messages if len(m) > 10]

# 数量
count = len(messages)

Agent实战：管理对话历史

from langchain_core.messages import HumanMessage, AIMessage

def manage_conversation(state: ChatState) -> ChatState:
    messages = state["messages"]
    
    # 保持最近10条消息（防止上下文太长）
    if len(messages) > 10:
        # 保留系统消息 + 最近9条
        system_msg = messages[0]
        recent_msgs = messages[-9:]
        messages = [system_msg] + recent_msgs
    
    # 添加新消息
    new_message = AIMessage(content="我的回复")
    messages.append(new_message)
    
    return {"messages": messages}

重点记忆：

• append()添加单个，extend()添加多个
• [-1]最后一个，[-3:]最后3个
• 列表推导式：[x for x in list if 条件]

---

4. 函数定义 - 节点的核心

为什么Agent开发需要：每个节点就是一个函数

基础函数

# 简单函数
def greet(name: str) -> str:
    return f"你好，{name}"

# 带默认参数
def greet(name: str, title: str = "先生") -> str:
    return f"你好，{title}{name}"

# 多返回值
def process() -> tuple[str, int]:
    return "结果", 100

result, count = process()

Agent实战：节点函数标准写法

from typing import TypedDict

class State(TypedDict):
    input: str
    output: str
    step: int

# 标准节点函数：接收State，返回State
def my_node(state: State) -> State:
    # 1. 从状态读取数据
    user_input = state["input"]
    
    # 2. 执行业务逻辑
    result = process_logic(user_input)
    
    # 3. 返回新状态（只返回改变的字段）
    return {
        "output": result,
        "step": state["step"] + 1
    }

# 条件函数：决定下一步
def route_condition(state: State) -> str:
    if state["step"] > 5:
        return "end"
    return "continue"

重点记忆：

• 节点函数：(state: State) -> State
• 路由函数：(state: State) -> str
• 只返回修改的字段即可

---

5. 条件语句 - 路由控制

为什么Agent开发需要：决定Agent的执行路径

基础语法

# if-elif-else
if score >= 90:
    result = "优秀"
elif score >= 60:
    result = "及格"
else:
    result = "不及格"

# 三元表达式（简洁写法）
result = "成功" if success else "失败"

# 多条件
if age > 18 and has_permission:
    print("允许")

if is_admin or is_vip:
    print("通过")

Agent实战：智能路由

from typing import Literal

def router(state: AgentState) -> Literal["tool", "respond", "human"]:
    """根据状态决定下一步"""
    
    # 获取最后一条消息
    last_message = state["messages"][-1]
    
    # 如果需要调用工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tool"
    
    # 如果是复杂问题，转人工
    if "复杂" in last_message.content or "转人工" in last_message.content:
        return "human"
    
    # 默认直接回复
    return "respond"

# 在图中使用
workflow.add_conditional_edges(
    "agent",
    router,
    {
        "tool": "tool_node",
        "respond": "respond_node",
        "human": "human_node"
    }
)

重点记忆：

• 路由函数返回字符串，决定下一个节点
• 用Literal["选项1", "选项2"]限定返回值
• hasattr(obj, "属性")检查对象是否有某属性

---

6. 异常处理 - 让Agent更稳定

为什么Agent开发需要：API调用可能失败，必须处理错误

基础语法

# 基本try-except
try:
    result = risky_operation()
except Exception as e:
    print(f"出错了: {e}")

# 多种异常
try:
    result = api_call()
except TimeoutError:
    print("超时")
except ConnectionError:
    print("连接失败")
except Exception as e:
    print(f"其他错误: {e}")

# finally总是执行
try:
    do_something()
except:
    handle_error()
finally:
    cleanup()  # 无论如何都执行

Agent实战：稳定的API调用

from typing import TypedDict

class AgentState(TypedDict):
    query: str
    result: str
    error: str
    retry_count: int

def call_llm_safely(state: AgentState) -> AgentState:
    """安全调用LLM，带重试和错误处理"""
    
    max_retries = 3
    retry_count = state.get("retry_count", 0)
    
    try:
        # 调用LLM
        from langchain_openai import ChatOpenAI
        llm = ChatOpenAI(model="gpt-3.5-turbo", timeout=30)
        
        response = llm.invoke(state["query"])
        
        return {
            "result": response.content,
            "error": "",
            "retry_count": 0
        }
        
    except TimeoutError:
        error_msg = "LLM调用超时"
        
    except Exception as e:
        error_msg = f"LLM调用失败: {str(e)}"
    
    # 处理错误
    if retry_count < max_retries:
        return {
            "error": error_msg,
            "retry_count": retry_count + 1
        }
    else:
        return {
            "result": "抱歉，服务暂时不可用",
            "error": f"重试{max_retries}次后仍失败",
            "retry_count": retry_count
        }

重点记忆：

• 永远不要让Agent因为异常而崩溃
• 用try-except包裹所有外部调用
• 记录错误信息到状态中

---

第二部分：常用进阶知识（⭐⭐⭐⭐）

7. 装饰器 - 定义Agent工具

为什么Agent开发需要：用@tool装饰器定义工具函数

基础概念

# 装饰器就是包装函数的函数
def my_decorator(func):
    def wrapper(*args, **kwargs):
        print("执行前")
        result = func(*args, **kwargs)
        print("执行后")
        return result
    return wrapper

@my_decorator
def say_hello():
    print("Hello")

# 相当于：say_hello = my_decorator(say_hello)

Agent实战：定义工具

from langchain_core.tools import tool

# 定义搜索工具
@tool
def search_web(query: str) -> str:
    """在网络上搜索信息
    
    Args:
        query: 搜索关键词
    
    Returns:
        搜索结果
    """
    # 这里调用真实的搜索API
    return f"搜索'{query}'的结果"

# 定义计算器工具
@tool
def calculate(expression: str) -> str:
    """计算数学表达式
    
    Args:
        expression: 数学表达式，如"2+3*4"
    
    Returns:
        计算结果
    """
    try:
        result = eval(expression)
        return f"结果是: {result}"
    except:
        return "计算错误"

# 工具列表
tools = [search_web, calculate]

# 绑定到LLM
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4").bind_tools(tools)

重点记忆：

• @tool装饰器让函数变成Agent工具
• 必须写docstring（LLM会读它来理解工具）
• Args和Returns要写清楚

---

8. 环境变量 - 管理API密钥

为什么Agent开发需要：API密钥不能写死在代码里

基础操作

import os

# 设置环境变量
os.environ["OPENAI_API_KEY"] = "sk-xxx"

# 读取环境变量
api_key = os.environ.get("OPENAI_API_KEY")
api_key = os.getenv("OPENAI_API_KEY", "默认值")

# 检查是否存在
if "OPENAI_API_KEY" in os.environ:
    print("已配置")

Agent实战：安全配置

# 方法1：使用.env文件（推荐）
# 1. 创建.env文件
"""
OPENAI_API_KEY=sk-your-key-here
ANTHROPIC_API_KEY=sk-ant-your-key
DATABASE_URL=postgresql://...
"""

# 2. 在代码中加载
from dotenv import load_dotenv
import os

load_dotenv()  # 自动加载.env文件

# 3. 使用
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4",
    api_key=os.getenv("OPENAI_API_KEY")  # 从环境变量读取
)

# 方法2：配置类（大项目推荐）
class Config:
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
    MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3"))
    TIMEOUT = int(os.getenv("TIMEOUT", "30"))
    
    @classmethod
    def validate(cls):
        """检查必需的配置"""
        if not cls.OPENAI_API_KEY:
            raise ValueError("请设置OPENAI_API_KEY环境变量")

# 使用
Config.validate()
llm = ChatOpenAI(api_key=Config.OPENAI_API_KEY)

重点记忆：

• 用.env文件管理密钥
• 用python-dotenv加载环境变量
• 永远不要把密钥提交到Git

---

9. JSON处理 - 数据交互

为什么Agent开发需要：LLM返回JSON，工具接收JSON

基础操作

import json

# Python对象 → JSON字符串
data = {"name": "张三", "age": 25}
json_str = json.dumps(data, ensure_ascii=False)  # 中文不转义

# JSON字符串 → Python对象
json_str = '{"name": "张三", "age": 25}'
data = json.loads(json_str)

# 读写文件
# 写入
with open("data.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

# 读取
with open("data.json", "r", encoding="utf-8") as f:
    data = json.load(f)

Agent实战：结构化输出

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import json

def extract_info(state: AgentState) -> AgentState:
    """从文本中提取结构化信息"""
    
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = f"""
    从以下文本中提取信息，返回JSON格式：
    {{
        "name": "姓名",
        "age": 年龄,
        "city": "城市"
    }}
    
    文本：{state["input"]}
    
    只返回JSON，不要其他内容。
    """
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    try:
        # 解析LLM返回的JSON
        info = json.loads(response.content)
        
        return {
            "extracted_info": info,
            "error": ""
        }
    except json.JSONDecodeError as e:
        return {
            "extracted_info": {},
            "error": f"JSON解析失败: {e}"
        }

# 使用Pydantic做结构化输出（更好的方法）
from pydantic import BaseModel

class PersonInfo(BaseModel):
    name: str
    age: int
    city: str

def extract_with_pydantic(text: str) -> PersonInfo:
    llm = ChatOpenAI(model="gpt-4")
    structured_llm = llm.with_structured_output(PersonInfo)
    
    result = structured_llm.invoke(f"提取信息: {text}")
    return result  # 自动是PersonInfo对象

重点记忆：

• json.dumps()和json.loads()用于转换
• ensure_ascii=False保留中文
• 用Pydantic做结构化输出更可靠

---

10. 字符串操作 - 处理文本

为什么Agent开发需要：处理用户输入、格式化输出

常用操作

text = "  Hello World  "

# 清理
text.strip()           # 去除首尾空格
text.lower()           # 转小写
text.upper()           # 转大写

# 查找
text.startswith("He")  # 是否以某字符串开头
text.endswith("ld")    # 是否以某字符串结尾
"World" in text        # 是否包含

# 分割和连接
parts = text.split()   # 按空格分割 → ["Hello", "World"]
text = "-".join(parts) # 用-连接 → "Hello-World"

# 替换
text.replace("Hello", "Hi")

# 格式化（推荐f-string）
name = "张三"
age = 25
message = f"{name}今年{age}岁"  # 推荐
message = "{}今年{}岁".format(name, age)

Agent实战：消息处理

def process_user_input(state: ChatState) -> ChatState:
    """清理和处理用户输入"""
    
    user_message = state["messages"][-1].content
    
    # 1. 清理输入
    cleaned = user_message.strip()
    
    # 2. 检测意图
    intent = "unknown"
    if any(word in cleaned.lower() for word in ["你好", "hello", "hi"]):
        intent = "greeting"
    elif any(word in cleaned.lower() for word in ["帮助", "help"]):
        intent = "help"
    elif "?" in cleaned or "？" in cleaned:
        intent = "question"
    
    # 3. 提取关键词
    keywords = [w for w in cleaned.split() if len(w) > 2]
    
    return {
        "intent": intent,
        "keywords": keywords,
        "cleaned_input": cleaned
    }

# 模板消息构建
def build_prompt(user_query: str, context: dict) -> str:
    """构建提示词模板"""
    
    template = f"""
你是一个智能助手。

用户信息：
- 姓名：{context.get('name', '未知')}
- 历史对话数：{len(context.get('history', []))}

用户问题：{user_query}

请根据上下文回答用户问题。
"""
    return template.strip()

重点记忆：

• 用f-string格式化字符串
• strip()去空格，lower()统一大小写
• in检查子串，split()分割

---

第三部分：选学知识（⭐⭐⭐）

11. 异步编程 - 提升性能

为什么Agent开发需要：并发调用多个API，提升速度

基础语法

import asyncio

# 定义异步函数
async def fetch_data(url: str) -> str:
    # 模拟网络请求
    await asyncio.sleep(1)
    return f"数据来自{url}"

# 运行异步函数
async def main():
    result = await fetch_data("https://api.com")
    print(result)

# 执行
asyncio.run(main())

# 并发执行多个任务
async def fetch_all():
    task1 = fetch_data("url1")
    task2 = fetch_data("url2")
    task3 = fetch_data("url3")
    
    # 并发执行，总耗时约1秒而不是3秒
    results = await asyncio.gather(task1, task2, task3)
    return results

Agent实战：并发工具调用

import asyncio
from typing import List

async def call_tool_async(tool_name: str, args: dict) -> str:
    """异步调用工具"""
    # 模拟工具调用
    await asyncio.sleep(1)
    return f"{tool_name}的结果"

async def parallel_tools(state: AgentState) -> AgentState:
    """并发调用多个工具"""
    
    # 需要调用的工具列表
    tool_calls = [
        ("search", {"query": "Python"}),
        ("calculate", {"expr": "2+2"}),
        ("translate", {"text": "Hello"})
    ]
    
    # 创建异步任务
    tasks = [
        call_tool_async(tool, args) 
        for tool, args in tool_calls
    ]
    
    # 并发执行（3秒变成1秒）
    results = await asyncio.gather(*tasks)
    
    return {
        "tool_results": results
    }

# LangGraph支持异步节点
from langgraph.graph import StateGraph

workflow = StateGraph(AgentState)
workflow.add_node("tools", parallel_tools)  # 自动识别异步函数

重点记忆：

• async def定义异步函数
• await等待异步操作
• asyncio.gather()并发执行多个任务
• LangGraph自动支持异步节点

---

12. 日志记录 - 调试必备

为什么Agent开发需要：追踪Agent执行过程

基础配置

import logging

# 基础配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

# 不同级别的日志
logger.debug("调试信息")
logger.info("一般信息")
logger.warning("警告")
logger.error("错误")
logger.critical("严重错误")

Agent实战：追踪执行

import logging
from typing import TypedDict

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('agent.log'),  # 写入文件
        logging.StreamHandler()             # 输出到控制台
    ]
)

logger = logging.getLogger(__name__)

class AgentState(TypedDict):
    step: int
    result: str

def logged_node(state: AgentState) -> AgentState:
    """带日志的节点"""
    
    step = state["step"]
    logger.info(f"开始执行步骤 {step}")
    
    try:
        # 执行业务逻辑
        result = process_data(state)
        
        logger.info(f"步骤 {step} 成功完成")
        logger.debug(f"结果: {result}")
        
        return {
            "step": step + 1,
            "result": result
        }
        
    except Exception as e:
        logger.error(f"步骤 {step} 失败: {e}", exc_info=True)
        
        return {
            "step": step,
            "result": f"错误: {e}"
        }

# 日志不同级别的使用场景
def agent_pipeline(state: AgentState):
    logger.debug(f"状态详情: {state}")           # 调试时看详细信息
    logger.info("开始处理用户请求")               # 正常流程记录
    logger.warning("API配额即将用完")            # 需要注意的情况
    logger.error("LLM调用失败，开始重试")        # 错误但可恢复
    logger.critical("数据库连接失败，系统停止")   # 严重错误

重点记忆：

• 开发时用DEBUG级别，生产用INFO或WARNING
• 用logger.exception()记录完整错误堆栈
• 日志同时输出到文件和控制台

---

第四部分：实战速查表

常用代码模板

1. 标准Agent节点

from typing import TypedDict

class State(TypedDict):
    input: str
    output: str

def my_node(state: State) -> State:
    """节点模板"""
    try:
        # 业务逻辑
        result = process(state["input"])
        return {"output": result}
    except Exception as e:
        logger.error(f"节点执行失败: {e}")
        return {"output": f"错误: {e}"}

2. 条件路由

from typing import Literal

def router(state: State) -> Literal["path1", "path2", "end"]:
    """路由模板"""
    if condition1:
        return "path1"
    elif condition2:
        return "path2"
    return "end"

3. 带重试的API调用

def call_api_with_retry(state: State) -> State:
    """带重试的API调用模板"""
    max_retries = 3
    
    for attempt in range(max_retries):
        try:
            result = api_call()
            return {"result": result, "error": ""}
        except Exception as e:
            if attempt == max_retries - 1:
                return {"result": "", "error": str(e)}
            time.sleep(2 ** attempt)  # 指数退避

4. LLM调用标准写法

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

def call_llm(state: State) -> State:
    """LLM调用模板"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    messages = [
        SystemMessage(content="你是一个助手"),
        HumanMessage(content=state["query"])
    ]
    
    response = llm.invoke(messages)
    
    return {"response": response.content}

5. 工具定义模板

from langchain_core.tools import tool

@tool
def my_tool(param1: str, param2: int) -> str:
    """工具的简短描述
    
    Args:
        param1: 参数1的说明
        param2: 参数2的说明
    
    Returns:
        返回值说明
    """
    # 实现逻辑
    return f"结果: {param1}, {param2}"

---

第五部分：避坑指南

❌ 常见错误

1. 忘记类型注解

# 错误
def node(state):  # 缺少类型注解
    return {"result": ""}

# 正确
def node(state: State) -> State:
    return {"result": ""}

2. 状态覆盖问题

# 错误：会覆盖整个列表
class State(TypedDict):
    messages: list

# 正确：会追加到列表
from typing import Annotated
from operator import add

class State(TypedDict):
    messages: Annotated[list, add]

3. 忘记处理异常

# 错误：API失败会导致整个Agent崩溃
def node(state: State):
    result = api_call()  # 可能失败
    return {"result": result}

# 正确
def node(state: State):
    try:
        result = api_call()
        return {"result": result}
    except Exception as e:
        return {"result": "", "error": str(e)}

4. 密钥硬编码

# 错误：密钥写在代码里
api_key = "sk-xxxxx"

# 正确：从环境变量读取
import os
api_key = os.getenv("OPENAI_API_KEY")

---

总结：学习路径

第1周：基础必会（⭐⭐⭐⭐⭐）

• [ ] 类型注解（TypedDict, Annotated）
• [ ] 字典操作（get, update, 合并）
• [ ] 列表操作（append, 切片, 推导式）
• [ ] 函数定义（参数、返回值、类型注解）
• [ ] 条件语句（if-else, 三元表达式）
• [ ] 异常处理（try-except-finally）

第2周：进阶常用（⭐⭐⭐⭐）

• [ ] 装饰器（@tool的使用）
• [ ] 环境变量（dotenv, os.getenv）
• [ ] JSON处理（dumps, loads）
• [ ] 字符串操作（format, f-string）

第3周：性能优化（⭐⭐⭐）

• [ ] 异步编程（async/await）
• [ ] 日志记录（logging）

练习建议

1. 每天敲一遍代码模板
2. 遇到问题先查速查表
3. 记不住就多用几次
4. 写注释帮助记忆

---

快速参考卡片

# 1. 导入常用库
from typing import TypedDict, Annotated, Literal
from operator import add
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
import os
import json
import logging

# 2. 定义状态
class State(TypedDict):
    messages: Annotated[list, add]
    data: dict

# 3. 定义节点
def node(state: State) -> State:
    return {"data": {}}

# 4. 定义路由
def route(state: State) -> Literal["next", "end"]:
    return "next"

# 5. 构建图
graph = StateGraph(State)
graph.add_node("node", node)
graph.set_entry_point("node")
graph.add_conditional_edges("node", route, {
    "next": "node",
    "end": END
})
app = graph.compile()

# 6. 运行
result = app.invoke({"messages": [], "data": {}})

掌握这些，你就可以开发80%的Agent应用了！ 🚀

小白学习版本。

AI Agent开发 - 像搭积木一样简单的Python知识

想象你在教一个从没编过程的朋友，用最简单的话讲清楚每个概念

---

开篇：什么是Agent？先理解这个

🤖 Agent就是一个会干活的机器人

想象你雇了一个助理，你说："帮我订机票"

普通程序（不是Agent）：

你说：帮我订机票
程序：好的！（然后卡住了，因为不知道去哪）

Agent（智能助理）：

你说：帮我订机票
Agent：好的！请问去哪里？什么时候出发？
你说：去北京，下周五
Agent：我帮你查了，有3个航班可选...
你说：选第一个
Agent：好的，帮你订好了！

看到区别了吗？Agent能：

1. 问你问题（互动）
2. 记住你说的话（记忆）
3. 自己查资料（使用工具）
4. 根据情况调整（智能决策）

---

第一课：什么是"状态"？就是记事本

📒 概念：状态就是Agent的记事本

想象你的Agent有一个笔记本，上面记着：

• 用户叫什么名字？
• 聊了什么内容？
• 现在在做什么步骤？
• 有没有出错？

生活例子：

你去餐厅点餐：

服务员的记事本（状态）：
- 桌号：8号桌
- 客人姓名：张三
- 已点菜品：宫保鸡丁、米饭
- 特殊要求：不要辣
- 状态：等待上菜

💻 Python代码怎么写这个记事本？

# 这是定义记事本的格式
from typing import TypedDict

class 餐厅记事本(TypedDict):
    桌号: int           # 整数类型（1, 2, 3...）
    客人姓名: str       # 文字类型（"张三"）
    已点菜品: list      # 列表类型（["菜1", "菜2"]）
    特殊要求: str       # 文字类型
    是否已上菜: bool    # 对错类型（True或False）

解释每一行：

• TypedDict：告诉Python"我要定义一个记事本格式"
• 桌号: int：记事本里有个格子叫"桌号"，只能填数字
• 客人姓名: str：有个格子叫"客人姓名"，只能填文字
• 已点菜品: list：有个格子叫"已点菜品"，是一个清单

🎯 为什么要定义格式？

就像表格一样，每一格都规定好了：

┌──────────┬──────────┐
│ 桌号（数字）│  8        │
├──────────┼──────────┤
│ 姓名（文字）│  张三     │
├──────────┼──────────┤
│ 菜品（清单）│  宫保鸡丁 │
│          │  米饭     │
└──────────┴──────────┘

如果不定义，就会乱：有人写桌号写成"八号桌"（应该是8），程序就不认识了。

🔧 Agent场景：聊天机器人的记事本

from typing import TypedDict, Annotated
from operator import add

# 定义聊天机器人的记事本
class 聊天记事本(TypedDict):
    用户名字: str                      # 用户叫什么
    聊天记录: Annotated[list, add]    # 聊了什么（自动追加）
    当前话题: str                      # 在聊什么话题
    
# 使用这个记事本
我的记事本 = {
    "用户名字": "小明",
    "聊天记录": ["你好", "我叫小明"],
    "当前话题": "自我介绍"
}

# 读记事本
print(我的记事本["用户名字"])  # 输出：小明

特别注意：Annotated[list, add]是什么意思？

# 普通list：会被覆盖
聊天记录: list
# 如果你写：{"聊天记录": ["新消息"]}
# 旧消息就没了！

# Annotated[list, add]：会自动追加
聊天记录: Annotated[list, add]  
# 如果你写：{"聊天记录": ["新消息"]}
# 会变成：["旧消息1", "旧消息2", "新消息"]  ✓

---

第二课：什么是"节点"？就是工作站

🏪 概念：节点就是做事的地方

想象一个快递流程：

【收件站】 → 【分拣站】 → 【派送站】
   ↓            ↓            ↓
  收快递       分类包裹     送到家

每个站（节点）都有自己的工作：

• 收件站：扫描快递单号，录入系统
• 分拣站：按地址分类
• 派送站：送到客户手里

💻 Python代码怎么写节点？

节点就是一个函数，格式是：

def 节点名字(状态):
    # 1. 读取状态（看记事本）
    # 2. 做一些事情
    # 3. 返回新的状态（更新记事本）
    return 新状态

🎯 实际例子：快递收件站

from typing import TypedDict

# 定义快递的记事本
class 快递状态(TypedDict):
    快递单号: str
    收件人: str
    当前位置: str
    状态: str

# 节点1：收件站
def 收件站(状态: 快递状态) -> 快递状态:
    """
    这个函数负责收件
    """
    # 1. 读取状态
    单号 = 状态["快递单号"]
    收件人 = 状态["收件人"]
    
    # 2. 做事情：扫描快递
    print(f"正在扫描快递 {单号}，收件人：{收件人}")
    
    # 3. 返回新状态（更新记事本）
    return {
        "快递单号": 单号,
        "收件人": 收件人,
        "当前位置": "收件站",
        "状态": "已收件"
    }

# 节点2：分拣站
def 分拣站(状态: 快递状态) -> 快递状态:
    print(f"正在分拣快递 {状态['快递单号']}")
    
    return {
        "当前位置": "分拣站",
        "状态": "已分拣"
    }

解释：

• def 收件站(状态: 快递状态)：定义一个名叫"收件站"的节点
• 状态: 快递状态：这个节点需要一个"快递状态"的记事本
• -> 快递状态：这个节点会返回一个"快递状态"记事本
• 函数里面就是具体做的事情

🔧 Agent场景：聊天节点

from typing import TypedDict

class 聊天状态(TypedDict):
    用户输入: str
    机器人回复: str

# 节点：回复用户
def 聊天节点(状态: 聊天状态) -> 聊天状态:
    """
    这个节点负责生成回复
    """
    # 1. 读状态：看用户说了什么
    用户说的话 = 状态["用户输入"]
    
    # 2. 做事情：根据用户输入生成回复
    if "你好" in 用户说的话:
        回复 = "你好！很高兴见到你"
    elif "天气" in 用户说的话:
        回复 = "今天天气不错哦"
    else:
        回复 = "我不太明白你的意思"
    
    # 3. 返回新状态
    return {
        "用户输入": 用户说的话,
        "机器人回复": 回复
    }

# 测试一下
结果 = 聊天节点({"用户输入": "你好", "机器人回复": ""})
print(结果["机器人回复"])  # 输出：你好！很高兴见到你

注意：你不需要返回所有字段，只返回改变的就行！

# 状态有3个字段
class 状态(TypedDict):
    字段A: str
    字段B: str  
    字段C: str

def 节点(状态):
    # 只改了字段B，只返回字段B就行
    return {"字段B": "新值"}
    
# LangGraph会自动合并：
# 旧状态：{"字段A": "A", "字段B": "B", "字段C": "C"}
# 你返回：{"字段B": "新B"}
# 最终变成：{"字段A": "A", "字段B": "新B", "字段C": "C"}

---

第三课：什么是"边"？就是路径

🛤️ 概念：边就是节点之间的连接

还是快递的例子：

【收件站】 ──→ 【分拣站】 ──→ 【派送站】

箭头就是"边"，告诉快递："收件完了去哪里？"

两种边

1. 普通边（固定路线）

就像地铁：

1号线：A站 → B站 → C站 → D站

每次都走同样的路线。

代码：

from langgraph.graph import StateGraph, END

图 = StateGraph(快递状态)

# 添加节点
图.add_node("收件", 收件站)
图.add_node("分拣", 分拣站)

# 添加普通边（固定路线）
图.add_edge("收件", "分拣")  # 收件后去分拣
图.add_edge("分拣", END)      # 分拣后结束

2. 条件边（智能路线）

就像导航软件：

你在家 → 【判断天气】
           ↓
        下雨 → 打车
        晴天 → 骑车

根据情况走不同的路。

代码：

# 判断函数：决定下一步去哪
def 判断天气(状态):
    """
    这个函数看天气，返回下一步去哪
    """
    if 状态["天气"] == "下雨":
        return "打车"
    else:
        return "骑车"

# 在图里使用条件边
图.add_conditional_edges(
    "判断",          # 从"判断"节点出发
    判断天气,        # 用这个函数决定
    {
        "打车": "打车节点",  # 如果函数返回"打车"，就去"打车节点"
        "骑车": "骑车节点"   # 如果函数返回"骑车"，就去"骑车节点"
    }
)

🎯 完整例子：判断年龄

from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal

# 状态
class 用户状态(TypedDict):
    年龄: int
    结果: str

# 节点1：检查年龄
def 检查节点(状态: 用户状态) -> 用户状态:
    """就是传递状态，什么都不做"""
    return 状态

# 节点2：成年人处理
def 成年人节点(状态: 用户状态) -> 用户状态:
    return {"结果": f"{状态['年龄']}岁，已成年"}

# 节点3：未成年人处理
def 未成年节点(状态: 用户状态) -> 用户状态:
    return {"结果": f"{状态['年龄']}岁，未成年"}

# 判断函数：决定去哪
def 判断年龄(状态: 用户状态) -> Literal["成年", "未成年"]:
    """
    Literal的意思是：只能返回"成年"或"未成年"这两个值
    """
    if 状态["年龄"] >= 18:
        return "成年"
    else:
        return "未成年"

# 构建图
图 = StateGraph(用户状态)

# 添加节点
图.add_node("检查", 检查节点)
图.add_node("成年", 成年人节点)
图.add_node("未成年", 未成年节点)

# 设置起点
图.set_entry_point("检查")

# 添加条件边
图.add_conditional_edges(
    "检查",     # 从检查节点出发
    判断年龄,   # 用判断年龄函数决定
    {
        "成年": "成年",      # 如果返回"成年"，去成年节点
        "未成年": "未成年"   # 如果返回"未成年"，去未成年节点
    }
)

# 两个分支都结束
图.add_edge("成年", END)
图.add_edge("未成年", END)

# 编译成可运行的程序
应用 = 图.compile()

# 测试
结果1 = 应用.invoke({"年龄": 20, "结果": ""})
print(结果1["结果"])  # 20岁，已成年

结果2 = 应用.invoke({"年龄": 15, "结果": ""})
print(结果2["结果"])  # 15岁，未成年

---

第四课：字典 - Agent的基础

📦 概念：字典就是一个储物柜

想象一个储物柜，每个格子有标签：

┌─────────────┬─────────┐
│ 姓名        │  张三   │
├─────────────┼─────────┤
│ 年龄        │  25     │
├─────────────┼─────────┤
│ 城市        │  北京   │
└─────────────┴─────────┘

💻 Python字典

# 创建字典（储物柜）
用户信息 = {
    "姓名": "张三",
    "年龄": 25,
    "城市": "北京"
}

# 读取值（看某个格子）
名字 = 用户信息["姓名"]     # 张三
年龄 = 用户信息["年龄"]     # 25

# 修改值（改某个格子）
用户信息["年龄"] = 26

# 添加新值（加新格子）
用户信息["职业"] = "程序员"

# 现在字典变成了：
# {
#     "姓名": "张三",
#     "年龄": 26,
#     "城市": "北京",
#     "职业": "程序员"
# }

🚨 重要：用get()方法更安全

# 问题：如果格子不存在会报错
电话 = 用户信息["电话"]  # ❌ 报错！因为没有"电话"这个格子

# 解决：用get()方法
电话 = 用户信息.get("电话")           # 返回None（空），不报错
电话 = 用户信息.get("电话", "无")     # 返回"无"（你设定的默认值）

🔧 Agent场景：管理用户信息

def 处理用户(状态):
    """处理用户信息的节点"""
    
    # 安全地读取信息（用get避免报错）
    姓名 = 状态.get("姓名", "访客")
    年龄 = 状态.get("年龄", 0)
    
    # 根据信息生成欢迎语
    if 年龄 > 0:
        欢迎语 = f"你好，{姓名}！你今年{年龄}岁"
    else:
        欢迎语 = f"你好，{姓名}！"
    
    # 返回新状态
    return {
        "欢迎语": 欢迎语
    }

合并字典（重要！）

Agent开发经常需要合并状态：

# 旧状态
旧状态 = {"姓名": "张三", "年龄": 25}

# 新信息
新信息 = {"年龄": 26, "城市": "北京"}

# 合并（用**符号）
合并后 = {**旧状态, **新信息}
# 结果：{"姓名": "张三", "年龄": 26, "城市": "北京"}
#      年龄被更新了，姓名保留，城市是新增的

---

第五课：列表 - 管理历史记录

📝 概念：列表就是一个清单

想象一个购物清单：

购物清单：
1. 苹果
2. 牛奶
3. 面包

💻 Python列表

# 创建列表
购物清单 = ["苹果", "牛奶", "面包"]

# 添加项目（在末尾加）
购物清单.append("鸡蛋")
# 现在是：["苹果", "牛奶", "面包", "鸡蛋"]

# 添加多个项目
购物清单.extend(["香蕉", "橙子"])
# 现在是：["苹果", "牛奶", "面包", "鸡蛋", "香蕉", "橙子"]

# 访问某一项（从0开始数）
第一项 = 购物清单[0]    # "苹果"
第二项 = 购物清单[1]    # "牛奶"

# 访问最后一项
最后一项 = 购物清单[-1]   # "橙子"
倒数第二 = 购物清单[-2]   # "香蕉"

# 获取最后3项
最后3项 = 购物清单[-3:]   # ["鸡蛋", "香蕉", "橙子"]

# 数一共有多少项
数量 = len(购物清单)      # 6

🎯 列表切片（获取一部分）

数字列表 = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

# 获取前3个
前三个 = 数字列表[:3]        # [0, 1, 2]

# 获取后3个
后三个 = 数字列表[-3:]       # [7, 8, 9]

# 获取中间的（从位置2到位置5）
中间的 = 数字列表[2:5]       # [2, 3, 4]

🔧 Agent场景：管理聊天历史

from typing import TypedDict

class 聊天状态(TypedDict):
    聊天历史: list

def 管理对话(状态: 聊天状态) -> 聊天状态:
    """
    管理聊天历史，保持最近10条
    """
    历史 = 状态["聊天历史"]
    
    # 添加新消息
    新消息 = "用户：今天天气怎么样？"
    历史.append(新消息)
    
    # 如果历史太长，只保留最近10条
    if len(历史) > 10:
        历史 = 历史[-10:]  # 只要最后10条
    
    return {
        "聊天历史": 历史
    }

遍历列表

任务列表 = ["洗衣服", "做饭", "打扫"]

# 方法1：直接遍历
for 任务 in 任务列表:
    print(f"正在：{任务}")

# 输出：
# 正在：洗衣服
# 正在：做饭
# 正在：打扫

# 方法2：带序号遍历
for 序号, 任务 in enumerate(任务列表):
    print(f"{序号+1}. {任务}")

# 输出：
# 1. 洗衣服
# 2. 做饭
# 3. 打扫

---

第六课：条件判断 - 让Agent做决定

🤔 概念：if就是"如果...那么..."

生活中的例子：

如果下雨，那么带伞
否则，不带伞

💻 Python的if语句

天气 = "下雨"

if 天气 == "下雨":
    print("带伞")
else:
    print("不带伞")

注意缩进！Python用缩进表示代码块：

if 条件:
    这行要缩进（属于if）
    这行也要缩进（属于if）
这行不缩进（不属于if）

多个条件

分数 = 85

if 分数 >= 90:
    print("优秀")
elif 分数 >= 80:      # elif = else if（否则如果）
    print("良好")
elif 分数 >= 60:
    print("及格")
else:
    print("不及格")

🎯 简写（三元表达式）

# 完整写法
年龄 = 20
if 年龄 >= 18:
    状态 = "成年"
else:
    状态 = "未成年"

# 简写（一行搞定）
状态 = "成年" if 年龄 >= 18 else "未成年"

多个条件组合

年龄 = 25
有驾照 = True

# and（并且）：两个都要满足
if 年龄 >= 18 and 有驾照:
    print("可以开车")

# or（或者）：满足一个就行
是会员 = True
是VIP = False

if 是会员 or 是VIP:
    print("享受折扣")

🔧 Agent场景：意图识别

def 识别意图(状态):
    """识别用户想做什么"""
    
    用户输入 = 状态["用户输入"]
    
    # 判断意图
    if "天气" in 用户输入:
        意图 = "查天气"
    elif "订票" in 用户输入 or "机票" in 用户输入:
        意图 = "订机票"
    elif "你好" in 用户输入 or "hi" in 用户输入:
        意图 = "打招呼"
    else:
        意图 = "不确定"
    
    return {
        "意图": 意图
    }

# 测试
结果 = 识别意图({"用户输入": "北京天气怎么样"})
print(结果["意图"])  # 查天气

---

第七课：异常处理 - 让程序不崩溃

🛡️ 概念：try就是"试一试"

生活例子：

试着用钥匙开门
如果钥匙不对：
    叫开锁师傅

为什么需要异常处理？

# 没有异常处理（危险！）
def 除法(a, b):
    return a / b

结果 = 除法(10, 0)  # ❌ 崩溃！不能除以0

程序直接崩溃，Agent就停止工作了。

💻 用try-except保护程序

def 安全除法(a, b):
    try:
        # 试着做这件事
        结果 = a / b
        return 结果
    except:
        # 如果出错了，做这个
        print("出错了！不能除以0")
        return None

结果 = 安全除法(10, 0)  # ✓ 不会崩溃，返回None

🎯 Agent必备：安全调用API

def 调用AI(状态):
    """安全地调用AI，不会因为网络问题崩溃"""
    
    try:
        # 尝试调用AI
        from langchain_openai import ChatOpenAI
        llm = ChatOpenAI(model="gpt-3.5-turbo")
        
        回复 = llm.invoke(状态["用户输入"])
        
        return {
            "AI回复": 回复.content,
            "错误": ""
        }
        
    except Exception as e:
        # 如果调用失败（网络问题、API错误等）
        return {
            "AI回复": "抱歉，我现在有点问题，请稍后再试",
            "错误": str(e)
        }

# 这样即使API调用失败，程序也不会崩溃

不同类型的错误

def 完整处理():
    try:
        # 做一些事
        pass
        
    except ZeroDivisionError:
        # 专门处理除以0的错误
        print("不能除以0")
        
    except KeyError:
        # 专门处理字典key不存在的错误
        print("这个key不存在")
        
    except Exception as e:
        # 处理所有其他错误
        print(f"发生了错误：{e}")
        
    finally:
        # 无论如何都会执行（通常用来清理资源）
        print("执行完毕")

---

第八课：把它们组合起来 - 完整例子

🎯 做一个简单的天气查询Agent

功能：用户说"北京天气"，Agent告诉他天气

流程图：

开始 → 理解输入 → 查天气 → 回复用户 → 结束

完整代码（每一行都有注释）

# 第1步：导入需要的工具
from langgraph.graph import StateGraph, END
from typing import TypedDict

# 第2步：定义状态（记事本）
class 天气Agent状态(TypedDict):
    用户输入: str      # 用户说的话
    城市: str          # 提取的城市名
    天气: str          # 查到的天气
    回复: str          # 最终回复

# 第3步：创建一个假的天气数据库
天气数据库 = {
    "北京": "晴天，15°C",
    "上海": "多云，18°C",
    "深圳": "小雨，25°C"
}

# 第4步：节点1 - 理解用户输入，提取城市
def 理解输入(状态: 天气Agent状态) -> 天气Agent状态:
    """从用户输入中找出城市名"""
    
    用户说的话 = 状态["用户输入"]
    
    # 简单判断（真实项目会用AI来理解）
    城市 = ""
    if "北京" in 用户说的话:
        城市 = "北京"
    elif "上海" in 用户说的话:
        城市 = "上海"
    elif "深圳" in 用户说的话:
        城市 = "深圳"
    
    return {"城市": 城市}

# 第5步：节点2 - 查询天气
def 查天气(状态: 天气Agent状态) -> 天气Agent状态:
    """根据城市查询天气"""
    
    城市 = 状态["城市"]
    
    # 从数据库查询
    天气 = 天气数据库.get(城市, "未找到该城市的天气")
    
    return {"天气": 天气}

# 第6步：节点3 - 生成回复
def 生成回复(状态: 天气Agent状态) -> 天气Agent状态:
    """根据查到的天气生成回复"""
    
    城市 = 状态["城市"]
    天气 = 状态["天气"]
    
    if 城市:
        回复 = f"{城市}的天气是：{天气}"
    else:
        回复 = "抱歉，我没有理解你想查哪个城市的天气"
    
    return {"回复": 回复}

# 第7步：构建图（连接所有节点）
天气图 = StateGraph(天气Agent状态)

# 添加3个节点
天气图.add_node("理解", 理解输入)
天气图.add_node("查询", 查天气)
天气图.add_node("回复", 生成回复)

# 设置起点
天气图.set_entry_point("理解")

# 连接节点（普通边）
天气图.add_edge("理解", "查询")   # 理解完去查询
天气图.add_edge("查询", "回复")   # 查询完去回复
天气图.add_edge("回复", END)      # 回复完结束

# 第8步：编译成可运行的程序
天气Agent = 天气图.compile()

# 第9步：测试
print("=== 测试1 ===")
结果1 = 天气Agent.invoke({
    "用户输入": "北京天气怎么样？",
    "城市": "",
    "天气": "",
    "回复": ""
})
print(结果1["回复"])  # 北京的天气是：晴天，15°C

print("\n=== 测试2 ===")
结果2 = 天气Agent.invoke({
    "用户输入": "上海今天下雨吗？",
    "城市": "",
    "天气": "",
    "回复": ""
})
print(结果2["回复"])  # 上海的天气是：多云，18°C

🎓 理解这个例子

1. 状态：就是一个记事本，记录4个信息
2. 节点：3个工作站，每个负责一件事
3. 边：用箭头连接，规定顺序
4. 运行：调用invoke()，传入初始状态

---

总结：你需要记住的核心

✅ 5个核心概念

1. 状态（State）：Agent的记事本，用TypedDict定义
2. 节点（Node）：做事的函数，格式是def 节点(状态) -> 状态
3. 边（Edge）：连接节点的路径，有普通边和条件边
4. 字典（Dict）：存储数据，用get()安全读取
5. 列表（List）：存储多个项目，用append()添加

✅ 学习路线

第1天：理解状态和节点

• 什么是状态？（记事本）
• 什么是节点？（工作站）
• 写一个简单的节点函数

第2天：理解边和图

• 什么是普通边？（固定路线）
• 什么是条件边？（智能路线）
• 把节点连起来

第3天：练习字典和列表

• 读写字典
• 操作列表
• 在节点中使用

第4天：学习异常处理

• 为什么需要try-except
• 保护API调用
• 让Agent更稳定

第5天：完整例子

• 跟着天气Agent例子敲一遍
• 修改代码，看看会发生什么
• 试着自己做一个

📝 每天练习清单

# 每天把这个模板写3遍，直到闭着眼睛都能写

from langgraph.graph import StateGraph, END
from typing import TypedDict

# 1. 定义状态
class MyState(TypedDict):
    data: str

# 2. 定义节点
def my_node(state: MyState) -> MyState:
    return {"data": "结果"}

# 3. 构建图
graph = StateGraph(MyState)
graph.add_node("node", my_node)
graph.set_entry_point("node")
graph.add_edge("node", END)

# 4. 运行
app = graph.compile()
result = app.invoke({"data": "输入"})

🎯 记不住？用这些口诀

1. 状态是记事本 - 记录所有信息
2. 节点是工作站 - 每个负责一件事
3. 边是路径 - 连接节点，规定顺序
4. 字典用get() - 安全读取，不怕报错
5. API加try - 保护程序，不会崩溃

---

遇到问题怎么办？

常见错误和解决方法

错误1：KeyError

# 错误写法
名字 = 状态["姓名"]  # 如果没有"姓名"这个key，会报错

# 正确写法
名字 = 状态.get("姓名", "默认值")  # 没有就用默认值

错误2：缩进错误

# 错误：缩进不对
def my_function():
print("hello")  # ❌ 没有缩进

# 正确
def my_function():
    print("hello")  # ✓ 缩进4个空格

错误3：忘记返回状态

# 错误
def node(state):
    结果 = "完成"
    # ❌ 忘记return了

# 正确
def node(state):
    结果 = "完成"
    return {"结果": 结果}  # ✓ 要返回

---

记住：学编程就像学骑自行车，刚开始会摔倒，多练几次就会了！

每天练习30分钟，1周后你就能写简单的Agent了！ 🚀

有任何不懂的，随时问我，我会用更简单的方式解释给你听！