2:从“模型中心“到“Agentic系统中心“：2026 AI技术栈全景图

业务问题 → 选择模型 → 设计Prompt → 获取输出 → 人工处理 → 业务价值

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor

# 定义状态类型
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next_step: str
    iteration_count: int
    max_iterations: int

# 定义Workflow节点
def planner_node(state: AgentState):
    """规划节点：分解任务并制定执行计划"""
    messages = state["messages"]
    
    planning_prompt = f"""
    基于以下对话历史，制定执行计划：
    {messages}
    
    请输出JSON格式的计划：
    {{
        "steps": ["步骤1", "步骤2", ...],
        "required_tools": ["工具1", "工具2"],
        "estimated_cost": 数字
    }}
    """
    
    response = llm.invoke(planning_prompt)
    plan = json.loads(response.content)
    
    return {
        "messages": [AIMessage(content=f"计划制定完成：{plan}")],
        "next_step": "execute"
    }

def executor_node(state: AgentState):
    """执行节点：调用工具或模型执行任务"""
    # 执行逻辑
    pass

def reflector_node(state: AgentState):
    """反思节点：评估执行结果并决定下一步"""
    messages = state["messages"]
    iteration = state["iteration_count"]
    max_iter = state["max_iterations"]
    
    if iteration >= max_iter:
        return {"next_step": END}
    
    reflection_prompt = f"""
    评估当前执行结果，决定是继续还是结束：
    {messages}
    
    如果任务完成，回复"COMPLETE"。
    如果需要继续，回复"CONTINUE"并说明原因。
    """
    
    response = llm.invoke(reflection_prompt)
    
    if "COMPLETE" in response.content:
        return {"next_step": END}
    else:
        return {
            "next_step": "planner",
            "iteration_count": iteration + 1
        }

# 构建Workflow
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("planner", planner_node)
workflow.add_node("executor", executor_node)
workflow.add_node("reflector", reflector_node)

# 添加边
workflow.set_entry_point("planner")
workflow.add_edge("planner", "executor")
workflow.add_edge("executor", "reflector")
workflow.add_conditional_edges(
    "reflector",
    lambda state: state["next_step"],
    {
        "planner": "planner",
        END: END
    }
)

# 编译
app = workflow.compile()

# 执行
result = app.invoke({
    "messages": [HumanMessage(content="分析2026年AI市场趋势")],
    "iteration_count": 0,
    "max_iterations": 5
})

from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
import numpy as np
from sentence_transformers import SentenceTransformer

@dataclass
class MemoryEntry:
    """记忆条目"""
    id: str
    content: str
    modality: str  # text, image, audio, video
    embedding: Optional[np.ndarray] = None
    timestamp: datetime = datetime.now()
    metadata: Dict[str, Any] = None
    importance_score: float = 0.5

class MultimodalMemory:
    """
    多模态记忆系统
    支持文本、图像、音频、视频的统一存储与检索
    """
    
    def __init__(self, embedding_model: str = "all-MiniLM-L6-v2"):
        self.short_term: List[MemoryEntry] = []
        self.long_term: List[MemoryEntry] = []
        self.structured: Dict[str, Any] = {}
        self.embedding_model = SentenceTransformer(embedding_model)
        self.max_short_term = 20
        
    def add(self, entry: MemoryEntry, to_long_term: bool = False):
        """
        添加记忆
        
        Args:
            entry: 记忆条目
            to_long_term: 是否直接存入长期记忆
        """
        # 生成embedding
        if entry.embedding is None and entry.modality == "text":
            entry.embedding = self.embedding_model.encode(entry.content)
        
        if to_long_term:
            self.long_term.append(entry)
        else:
            self.short_term.append(entry)
            # 短期记忆溢出时，转移到长期记忆
            if len(self.short_term) > self.max_short_term:
                self._consolidate_to_long_term()
    
    def retrieve(
        self, 
        query: str, 
        k: int = 5,
        memory_type: str = "all"  # short, long, all
    ) -> List[MemoryEntry]:
        """
        检索相关记忆
        
        Args:
            query: 查询文本
            k: 返回条目数
            memory_type: 记忆类型筛选
            
        Returns:
            相关记忆条目列表
        """
        query_embedding = self.embedding_model.encode(query)
        
        candidates = []
        
        if memory_type in ["short", "all"]:
            candidates.extend(self.short_term)
        if memory_type in ["long", "all"]:
            candidates.extend(self.long_term)
        
        # 计算相似度
        scored = []
        for entry in candidates:
            if entry.embedding is not None:
                similarity = np.dot(query_embedding, entry.embedding)
                # 重要性加权
                weighted_score = similarity * entry.importance_score
                scored.append((entry, weighted_score))
        
        # 按分数排序并返回Top K
        scored.sort(key=lambda x: x[1], reverse=True)
        return [entry for entry, _ in scored[:k]]
    
    def _consolidate_to_long_term(self):
        """
        将短期记忆整合到长期记忆
        基于重要性评分进行筛选
        """
        # 按重要性排序
        self.short_term.sort(key=lambda x: x.importance_score, reverse=True)
        
        # 保留高重要性记忆到长期记忆
        for entry in self.short_term[self.max_short_term//2:]:
            if entry.importance_score > 0.7:
                self.long_term.append(entry)
        
        # 清空短期记忆（保留最近的一半）
        self.short_term = self.short_term[:self.max_short_term//2]
    
    def get_context_for_llm(self, max_tokens: int = 4000) -> str:
        """
        获取适合LLM上下文的记忆摘要
        
        Args:
            max_tokens: 最大token数
            
        Returns:
            格式化的记忆文本
        """
        context_parts = []
        
        # 添加结构化记忆
        if self.structured:
            context_parts.append("## 已知信息")
            for key, value in self.structured.items():
                context_parts.append(f"- {key}: {value}")
        
        # 添加短期记忆
        if self.short_term:
            context_parts.append("\n## 近期对话")
            for entry in self.short_term[-5:]:
                context_parts.append(f"[{entry.timestamp}] {entry.content}")
        
        return "\n".join(context_parts)

# 使用示例
memory = MultimodalMemory()

# 添加记忆
memory.add(MemoryEntry(
    id="conv_001",
    content="用户偏好使用Python进行数据分析",
    modality="text",
    importance_score=0.9
), to_long_term=True)

# 检索记忆
results = memory.retrieve("用户喜欢什么编程语言？", k=3)

from typing import List, Dict, Tuple, Set
import networkx as nx
from sentence_transformers import SentenceTransformer

class GraphRAGRouter:
    """
    GraphRAG Router
    结合向量检索和知识图谱的智能路由系统
    """
    
    def __init__(self):
        self.graph = nx.DiGraph()
        self.embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
        self.node_embeddings: Dict[str, np.ndarray] = {}
        
    def add_knowledge(
        self, 
        entities: List[str], 
        relations: List[Tuple[str, str, str]],
        documents: List[str]
    ):
        """
        添加知识到图谱
        
        Args:
            entities: 实体列表
            relations: 关系三元组 (subject, predicate, object)
            documents: 相关文档
        """
        # 添加实体节点
        for entity in entities:
            self.graph.add_node(entity, type="entity")
            self.node_embeddings[entity] = self.embedding_model.encode(entity)
        
        # 添加关系边
        for subj, pred, obj in relations:
            self.graph.add_edge(subj, obj, relation=pred)
        
        # 添加文档节点
        for i, doc in enumerate(documents):
            doc_id = f"doc_{i}"
            self.graph.add_node(doc_id, type="document", content=doc)
            self.node_embeddings[doc_id] = self.embedding_model.encode(doc)
    
    def retrieve(
        self, 
        query: str, 
        k: int = 5,
        max_hops: int = 2
    ) -> Dict[str, Any]:
        """
        检索相关知识
        
        Args:
            query: 查询文本
            k: 返回实体数
            max_hops: 最大图遍历深度
            
        Returns:
            包含检索结果和推理路径的字典
        """
        query_embedding = self.embedding_model.encode(query)
        
        # Step 1: 向量检索找到入口节点
        entry_nodes = self._vector_search(query_embedding, k=3)
        
        # Step 2: 图遍历扩展相关节点
        expanded_nodes = self._graph_expand(entry_nodes, max_hops)
        
        # Step 3: 重新排序
        ranked_results = self._rerank(query_embedding, expanded_nodes)
        
        # Step 4: 构建推理路径
        paths = self._extract_paths(entry_nodes, ranked_results[:k])
        
        return {
            "results": ranked_results[:k],
            "entry_points": entry_nodes,
            "reasoning_paths": paths,
            "coverage_score": self._calculate_coverage(query, ranked_results[:k])
        }
    
    def _vector_search(
        self, 
        query_embedding: np.ndarray, 
        k: int
    ) -> List[str]:
        """向量相似度搜索"""
        scores = []
        for node_id, embedding in self.node_embeddings.items():
            similarity = np.dot(query_embedding, embedding)
            scores.append((node_id, similarity))
        
        scores.sort(key=lambda x: x[1], reverse=True)
        return [node_id for node_id, _ in scores[:k]]
    
    def _graph_expand(
        self, 
        seed_nodes: List[str], 
        max_hops: int
    ) -> Set[str]:
        """图遍历扩展"""
        expanded = set(seed_nodes)
        current_layer = set(seed_nodes)
        
        for _ in range(max_hops):
            next_layer = set()
            for node in current_layer:
                if node in self.graph:
                    neighbors = set(self.graph.neighbors(node))
                    predecessors = set(self.graph.predecessors(node))
                    next_layer.update(neighbors)
                    next_layer.update(predecessors)
            
            expanded.update(next_layer)
            current_layer = next_layer
        
        return expanded
    
    def _rerank(
        self, 
        query_embedding: np.ndarray, 
        candidates: Set[str]
    ) -> List[Dict]:
        """重新排序候选节点"""
        scored = []
        
        for node_id in candidates:
            if node_id in self.node_embeddings:
                # 向量相似度
                vec_score = np.dot(query_embedding, self.node_embeddings[node_id])
                
                # 图中心性
                centrality = nx.degree_centrality(self.graph).get(node_id, 0)
                
                # 综合评分
                final_score = 0.7 * vec_score + 0.3 * centrality
                
                node_data = self.graph.nodes[node_id]
                scored.append({
                    "id": node_id,
                    "score": final_score,
                    "type": node_data.get("type", "unknown"),
                    "content": node_data.get("content", node_id)
                })
        
        scored.sort(key=lambda x: x["score"], reverse=True)
        return scored
    
    def _extract_paths(
        self, 
        entry_nodes: List[str], 
        results: List[Dict]
    ) -> List[List[str]]:
        """提取从入口到结果的推理路径"""
        paths = []
        
        for entry in entry_nodes:
            for result in results:
                try:
                    path = nx.shortest_path(
                        self.graph, 
                        source=entry, 
                        target=result["id"]
                    )
                    paths.append(path)
                except nx.NetworkXNoPath:
                    continue
        
        return paths
    
    def _calculate_coverage(
        self, 
        query: str, 
        results: List[Dict]
    ) -> float:
        """计算检索覆盖率"""
        # 简化的覆盖率计算
        query_terms = set(query.lower().split())
        covered_terms = set()
        
        for result in results:
            content = result.get("content", "").lower()
            covered_terms.update(query_terms.intersection(content.split()))
        
        return len(covered_terms) / len(query_terms) if query_terms else 0.0

# 使用示例
router = GraphRAGRouter()

# 添加知识
router.add_knowledge(
    entities=["Agentic AI", "LLM", "RAG", "Workflow"],
    relations=[
        ("Agentic AI", "uses", "LLM"),
        ("Agentic AI", "uses", "RAG"),
        ("Agentic AI", "implements", "Workflow"),
        ("RAG", "enhances", "LLM")
    ],
    documents=[
        "Agentic AI systems use LLMs as core reasoning engines",
        "RAG enhances LLM capabilities with external knowledge"
    ]
)

# 检索
results = router.retrieve("How does Agentic AI work?", k=3)

from typing import Dict, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum

class TaskComplexity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class ModelConfig:
    name: str
    cost_per_1k_tokens: float
    avg_latency_ms: int
    capabilities: List[str]
    context_window: int

class SmartModelRouter:
    """
    智能模型路由器
    根据任务特征动态选择最优模型
    """
    
    def __init__(self):
        self.models: Dict[str, ModelConfig] = {
            "llama-4-8b": ModelConfig(
                name="Llama 4 8B",
                cost_per_1k_tokens=0.0001,
                avg_latency_ms=80,
                capabilities=["text", "simple_qa"],
                context_window=128000
            ),
            "gpt-5.4-mini": ModelConfig(
                name="GPT-5.4 Mini",
                cost_per_1k_tokens=0.0005,
                avg_latency_ms=150,
                capabilities=["text", "code", "reasoning"],
                context_window=256000
            ),
            "claude-opus-4.6": ModelConfig(
                name="Claude Opus 4.6",
                cost_per_1k_tokens=0.003,
                avg_latency_ms=400,
                capabilities=["text", "code", "reasoning", "analysis"],
                context_window=200000
            ),
            "gpt-5.4": ModelConfig(
                name="GPT-5.4",
                cost_per_1k_tokens=0.005,
                avg_latency_ms=600,
                capabilities=["text", "code", "reasoning", "multimodal", "analysis"],
                context_window=1000000
            ),
            "grok-4-moe": ModelConfig(
                name="Grok 4 MoE",
                cost_per_1k_tokens=0.004,
                avg_latency_ms=350,
                capabilities=["text", "code", "reasoning", "multimodal"],
                context_window=500000
            )
        }
        
        self.cost_budget = float("inf")
        self.latency_budget_ms = 1000
        
    def route(
        self, 
        task_description: str,
        required_capabilities: List[str],
        priority: str = "balanced"  # cost, quality, speed, balanced
    ) -> ModelConfig:
        """
        路由到最优模型
        
        Args:
            task_description: 任务描述
            required_capabilities: 必需能力列表
            priority: 优先级（成本、质量、速度、平衡）
            
        Returns:
            选中的模型配置
        """
        # 筛选满足能力要求的模型
        candidates = [
            model for model in self.models.values()
            if all(cap in model.capabilities for cap in required_capabilities)
        ]
        
        if not candidates:
            raise ValueError(f"No model supports all required capabilities: {required_capabilities}")
        
        # 根据优先级排序
        if priority == "cost":
            candidates.sort(key=lambda m: m.cost_per_1k_tokens)
        elif priority == "speed":
            candidates.sort(key=lambda m: m.avg_latency_ms)
        elif priority == "quality":
            # 质量与context window和延迟相关（通常大模型质量更高）
            candidates.sort(key=lambda m: m.context_window, reverse=True)
        else:  # balanced
            candidates.sort(key=lambda m: self._balanced_score(m))
        
        return candidates[0]
    
    def _balanced_score(self, model: ModelConfig) -> float:
        """计算综合评分（越低越好）"""
        # 归一化成本（假设最高成本0.01）
        cost_score = model.cost_per_1k_tokens / 0.01
        
        # 归一化延迟（假设最高延迟1000ms）
        latency_score = model.avg_latency_ms / 1000
        
        # 能力评分（能力越多分越高）
        capability_score = 1 - (len(model.capabilities) / 10)
        
        # 加权综合
        return 0.4 * cost_score + 0.4 * latency_score + 0.2 * capability_score
    
    def estimate_cost(
        self, 
        model_name: str, 
        input_tokens: int, 
        output_tokens: int
    ) -> float:
        """
        估算调用成本
        
        Args:
            model_name: 模型名称
            input_tokens: 输入token数
            output_tokens: 输出token数
            
        Returns:
            预估成本（美元）
        """
        if model_name not in self.models:
            raise ValueError(f"Unknown model: {model_name}")
        
        model = self.models[model_name]
        total_tokens = input_tokens + output_tokens
        
        return (total_tokens / 1000) * model.cost_per_1k_tokens
    
    def get_fallback_chain(
        self, 
        primary_model: str
    ) -> List[ModelConfig]:
        """
        获取降级链
        
        Args:
            primary_model: 主模型名称
            
        Returns:
            按优先级排序的降级模型列表
        """
        if primary_model not in self.models:
            return []
        
        primary = self.models[primary_model]
        
        # 找能力相似但成本更低的模型
        fallbacks = [
            model for name, model in self.models.items()
            if name != primary_model
            and set(model.capabilities) >= set(primary.capabilities)
            and model.cost_per_1k_tokens < primary.cost_per_1k_tokens
        ]
        
        fallbacks.sort(key=lambda m: m.cost_per_1k_tokens)
        return fallbacks

# 使用示例
router = SmartModelRouter()

# 路由任务
model = router.route(
    task_description="Generate Python code for data analysis",
    required_capabilities=["code", "reasoning"],
    priority="balanced"
)

print(f"Selected model: {model.name}")

# 估算成本
cost = router.estimate_cost(model.name, input_tokens=2000, output_tokens=500)
print(f"Estimated cost: ${cost:.4f}")

from typing import List, Dict, Any, Callable
from dataclasses import dataclass
import random

@dataclass
class SyntheticDataSample:
    """合成数据样本"""
    input_data: Any
    expected_output: Any
    metadata: Dict[str, Any]
    quality_score: float
    generation_method: str

class SyntheticDataGenerator:
    """
    合成数据生成器
    支持多种生成策略：模板填充、LLM生成、对抗生成、变异生成
    """
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.generation_strategies: Dict[str, Callable] = {
            "template": self._template_generation,
            "llm": self._llm_generation,
            "variation": self._variation_generation,
            "adversarial": self._adversarial_generation
        }
        
    def generate(
        self,
        task_type: str,
        num_samples: int,
        strategies: List[str] = None,
        seed_data: List[Dict] = None
    ) -> List[SyntheticDataSample]:
        """
        生成合成数据
        
        Args:
            task_type: 任务类型
            num_samples: 样本数量
            strategies: 生成策略列表
            seed_data: 种子数据
            
        Returns:
            合成数据样本列表
        """
        if strategies is None:
            strategies = ["template", "llm"]
        
        samples = []
        samples_per_strategy = num_samples // len(strategies)
        
        for strategy in strategies:
            if strategy in self.generation_strategies:
                strategy_samples = self.generation_strategies[strategy](
                    task_type=task_type,
                    num_samples=samples_per_strategy,
                    seed_data=seed_data
                )
                samples.extend(strategy_samples)
        
        # 质量过滤
        filtered_samples = [s for s in samples if s.quality_score > 0.7]
        
        return filtered_samples[:num_samples]
    
    def _template_generation(
        self, 
        task_type: str, 
        num_samples: int,
        seed_data: List[Dict]
    ) -> List[SyntheticDataSample]:
        """基于模板的生成"""
        templates = self._get_templates(task_type)
        samples = []
        
        for _ in range(num_samples):
            template = random.choice(templates)
            filled = self._fill_template(template)
            
            sample = SyntheticDataSample(
                input_data=filled["input"],
                expected_output=filled["output"],
                metadata={"template_id": template["id"]},
                quality_score=0.8,  # 模板填充质量较高
                generation_method="template"
            )
            samples.append(sample)
        
        return samples
    
    def _llm_generation(
        self, 
        task_type: str, 
        num_samples: int,
        seed_data: List[Dict]
    ) -> List[SyntheticDataSample]:
        """基于LLM的生成"""
        samples = []
        
        generation_prompt = f"""
        Generate {num_samples} synthetic training samples for task type: {task_type}.
        
        Format each sample as:
        Input: [input data]
        Output: [expected output]
        
        Ensure diversity and coverage of edge cases.
        """
        
        response = self.llm.chat.completions.create(
            model="gpt-5.4-mini",
            messages=[{"role": "user", "content": generation_prompt}]
        )
        
        # 解析生成的样本
        generated_text = response.choices[0].message.content
        parsed_samples = self._parse_generated_samples(generated_text)
        
        for parsed in parsed_samples:
            sample = SyntheticDataSample(
                input_data=parsed["input"],
                expected_output=parsed["output"],
                metadata={},
                quality_score=self._evaluate_quality(parsed),
                generation_method="llm"
            )
            samples.append(sample)
        
        return samples
    
    def _variation_generation(
        self, 
        task_type: str, 
        num_samples: int,
        seed_data: List[Dict]
    ) -> List[SyntheticDataSample]:
        """基于变异的生成"""
        if not seed_data:
            return []
        
        samples = []
        
        for _ in range(num_samples):
            seed = random.choice(seed_data)
            
            variation_prompt = f"""
            Create a variation of the following sample:
            Original Input: {seed['input']}
            Original Output: {seed['output']}
            
            Generate a similar but different sample with the same underlying pattern.
            """
            
            response = self.llm.chat.completions.create(
                model="gpt-5.4-mini",
                messages=[{"role": "user", "content": variation_prompt}]
            )
            
            varied = self._parse_variation(response.choices[0].message.content)
            
            sample = SyntheticDataSample(
                input_data=varied["input"],
                expected_output=varied["output"],
                metadata={"seed_id": seed.get("id")},
                quality_score=0.75,
                generation_method="variation"
            )
            samples.append(sample)
        
        return samples
    
    def _adversarial_generation(
        self, 
        task_type: str, 
        num_samples: int,
        seed_data: List[Dict]
    ) -> List[SyntheticDataSample]:
        """对抗性样本生成"""
        samples = []
        
        adversarial_prompt = f"""
        Generate adversarial test cases for task type: {task_type}.
        
        These should be:
        1. Semantically valid but potentially confusing
        2. Edge cases that might break the system
        3. Ambiguous inputs requiring careful handling
        
        Format: Input | Expected Behavior
        """
        
        response = self.llm.chat.completions.create(
            model="claude-opus-4.6",
            messages=[{"role": "user", "content": adversarial_prompt}]
        )
        
        # 解析对抗样本
        adversarial_samples = self._parse_adversarial(response.choices[0].message.content)
        
        for adv in adversarial_samples[:num_samples]:
            sample = SyntheticDataSample(
                input_data=adv["input"],
                expected_output=adv["expected_behavior"],
                metadata={"type": "adversarial"},
                quality_score=0.9,  # 对抗样本通常质量较高
                generation_method="adversarial"
            )
            samples.append(sample)
        
        return samples
    
    def _get_templates(self, task_type: str) -> List[Dict]:
        """获取任务模板"""
        # 模板库
        templates = {
            "qa": [
                {"id": "qa_1", "input_template": "What is {topic}?", "output_template": "{topic} is defined as..."},
                {"id": "qa_2", "input_template": "Explain {concept} in simple terms", "output_template": "{concept} can be understood as..."}
            ],
            "code": [
                {"id": "code_1", "input_template": "Write a function to {operation}", "output_template": "def {function_name}():..."}
            ]
        }
        return templates.get(task_type, [])
    
    def _fill_template(self, template: Dict) -> Dict:
        """填充模板"""
        # 简化实现
        return {
            "input": template.get("input_template", "").format(topic="AI"),
            "output": template.get("output_template", "").format(topic="AI")
        }
    
    def _parse_generated_samples(self, text: str) -> List[Dict]:
        """解析生成的样本"""
        # 简化实现
        return [{"input": text[:100], "output": text[100:200]}]
    
    def _parse_variation(self, text: str) -> Dict:
        """解析变异样本"""
        return {"input": text, "output": "varied output"}
    
    def _parse_adversarial(self, text: str) -> List[Dict]:
        """解析对抗样本"""
        return [{"input": text[:50], "expected_behavior": "handle carefully"}]
    
    def _evaluate_quality(self, sample: Dict) -> float:
        """评估样本质量"""
        # 基于启发式规则的质量评估
        score = 0.5
        
        if len(sample.get("input", "")) > 10:
            score += 0.2
        
        if len(sample.get("output", "")) > 20:
            score += 0.2
        
        return min(score, 1.0)

# 使用示例
generator = SyntheticDataGenerator(llm_client=openai.OpenAI())

samples = generator.generate(
    task_type="qa",
    num_samples=100,
    strategies=["template", "llm", "variation"]
)

print(f"Generated {len(samples)} synthetic samples")

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
import operator

class ResearchState(TypedDict):
    topic: str
    search_results: list
    extracted_info: dict
    report: str
    iteration: int

class ResearchAssistant:
    def __init__(self):
        self.workflow = self._build_workflow()
    
    def _build_workflow(self):
        workflow = StateGraph(ResearchState)
        
        # 添加节点
        workflow.add_node("search", self._search_node)
        workflow.add_node("extract", self._extract_node)
        workflow.add_node("generate", self._generate_node)
        workflow.add_node("review", self._review_node)
        
        # 设置边
        workflow.set_entry_point("search")
        workflow.add_edge("search", "extract")
        workflow.add_edge("extract", "generate")
        workflow.add_edge("generate", "review")
        workflow.add_conditional_edges(
            "review",
            self._should_continue,
            {"continue": "search", "end": END}
        )
        
        return workflow.compile()
    
    def run(self, topic: str):
        return self.workflow.invoke({
            "topic": topic,
            "search_results": [],
            "extracted_info": {},
            "report": "",
            "iteration": 0
        })
    
    def _search_node(self, state: ResearchState):
        # 实现搜索逻辑
        pass
    
    def _extract_node(self, state: ResearchState):
        # 实现信息提取逻辑
        pass
    
    def _generate_node(self, state: ResearchState):
        # 实现报告生成逻辑
        pass
    
    def _review_node(self, state: ResearchState):
        # 实现报告审核逻辑
        pass
    
    def _should_continue(self, state: ResearchState):
        if state["iteration"] < 3:
            return "continue"
        return "end"

# 运行
assistant = ResearchAssistant()
result = assistant.run("Agentic AI trends 2026")