LangChain 第四章 graph并行推理

LLM 有时可以同时处理一项任务，并以编程方式聚合其输出。这种工作流（并行化）表现为两个关键变体：分段：将任务分解为并行运行的独立子任务。投票：多次运行同一任务以获得不同的输出。

何时使用此工作流程：当可以并行化已划分的子任务以提高速度时，或者当需要多个视角或尝试以获得更高的置信度结果时，并行化是有效的。对于具有多个考虑因素的复杂任务，当每个考虑因素都由单独的 LLM 调用处理时，LLM 通常会表现得更好，从而可以将注意力集中在每个特定方面。

from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

OPENAI_API_KEY= "lm-studio"
OPENAI_MODEL_NAME = "qwen2.5-3b-instruct"
OPENAI_BASE_URL="http://localhost:9999/v1"
llm = ChatOpenAI(
    openai_api_key=OPENAI_API_KEY,
    model_name=OPENAI_MODEL_NAME,
    openai_api_base=OPENAI_BASE_URL,
    temperature=5
)
# Graph state


class State(TypedDict):
    topic: str
    joke: str
    story: str
    poem: str
    combined_output: str


# Nodes
def call_llm_1(state: State):
    """First LLM call to generate initial joke"""
    print(f"call_llm_1被调用，state->{state}")
    msg = llm.invoke(f"写一段关于 {state['topic']}的笑话")
    return {"joke": msg.content}


def call_llm_2(state: State):
    """Second LLM call to generate story"""
    print(f"call_llm_2被调用，state->{state}")
    msg = llm.invoke(f"写一段关于 {state['topic']}的故事")
    return {"story": msg.content}


def call_llm_3(state: State):
    """Third LLM call to generate poem"""
    print(f"call_llm_3被调用，state->{state}")
    msg = llm.invoke(f"写一首有关 {state['topic']}的7句诗")
    return {"poem": msg.content}


def aggregator(state: State):
    """Combine the joke and story into a single output"""

    combined = f"这里是关于 {state['topic']}的故事，笑话，诗歌!\n\n"
    combined += f"STORY:\n{state['story']}\n\n"
    combined += f"JOKE:\n{state['joke']}\n\n"
    combined += f"POEM:\n{state['poem']}"
    return {"combined_output": combined}


# Build workflow
parallel_builder = StateGraph(State)

# Add nodes
parallel_builder.add_node("call_llm_1", call_llm_1)
parallel_builder.add_node("call_llm_2", call_llm_2)
parallel_builder.add_node("call_llm_3", call_llm_3)
parallel_builder.add_node("aggregator", aggregator)

# Add edges to connect nodes
parallel_builder.add_edge(START, "call_llm_1")
parallel_builder.add_edge(START, "call_llm_2")
parallel_builder.add_edge(START, "call_llm_3")
parallel_builder.add_edge("call_llm_1", "aggregator")
parallel_builder.add_edge("call_llm_2", "aggregator")
parallel_builder.add_edge("call_llm_3", "aggregator")
parallel_builder.add_edge("aggregator", END)
parallel_workflow = parallel_builder.compile()

# Show workflow

# Invoke
state = parallel_workflow.invoke({"topic": "喵咪"})
print(state["combined_output"])

输出结果：

call_llm_1被调用，state->{'topic': '喵咪'}
call_llm_2被调用，state->{'topic': '喵咪'}
call_llm_3被调用，state->{'topic': '喵咪'}
这里是关于 喵咪的故事，笑话，诗歌!

STORY:
喵喵是个非常有好奇心和活泼开朗的猫咪，它有着一身雪白如棉、光泽闪耀的毛发。每当微风吹过，仿佛那洁白柔软的毛发就轻轻跳跃着舞动一般。

在小猫成长的日子里，喵喵常常好奇地探索它从未见识过的角落，不论是家里的每一扇窗户边、每一个隐蔽的小缝隙里都留下了它探索的身影和脚印。在这个过程中，猫咪总会用它的眼睛来捕捉远处或隐藏的秘密，每当找到一些小虫子或其他动物时，总会兴奋地发出喵喵喵的叫声。

虽然在与猫科伙伴互动的时候有时候会显得比较凶悍和难以相处，但只要家人或是主人能给以温柔关怀，它便会乖乖地蜷缩在人怀里撒娇、依恋。每当这个时候喵喵的瞳孔会变得大大的充满暖意，在主人怀中眯眼轻喘。

喵喵最喜欢的就是玩小球的游戏了，在小猫成长的过程中它学会了用脚来追这些小小玩具，并且有时会在玩耍时不小心摔下来，却仍旧顽强地跳起来再追。尽管常常跌倒、摔跤，但它并不介意。反而更加快乐。

有时候会看到它静静地趴着观察窗外飞过的鸟儿或是一朵飘落的树叶，在这一刻它便如同一匹沉稳而威猛的猎豹一般静静潜伏。

总之喵喵就是个有生命力，爱玩又有点淘气的小猫哦。

JOKE:
喵星人的幽默真是难以捉摸啊，但还是让我们脑补一下这搞笑一幕吧： 

一个喵星人向它的邻居打招呼说：“喂，你家有吃不完的猫粮吗？”邻居惊讶地看着它说，“当然有的啦！我家每天都会多出一大罐。”喵星人大吃一惊：“那你得赶紧跟我搬家！”邻居一脸不解地说：“我还没吃过猫粮呢。”

这样的故事可能只能在脑内和喜欢猫咪的朋友们一起开开心心地乐呵一下，希望你也会被这诙谐的故事逗笑！

POEM:
猫咪身轻似月光，夜幕深处露欢笑。
黑白双纹织岁月，温馨故事伴晨晓。
温柔瞳中藏乾坤，捕鼠本领谁堪对？
世间琐碎皆化尘，唯君情深胜烟霞。