langchain

调用顺序：

• self.__call__() –> self.invoke() –> self._call
• self.acall() –> self.ainvoke() –> self._acall

Runnable

lanchain_core.runnables.base.Runnable

许多功能的基础类

invoke()

input -> output

batch()

a list of input -> a list of output, parallel 平行执行

stream()

input -> stream output

RunnableSequence

lanchain_core.runnables.base.RunnableSequence

RunnableParallel

lanchain_core.runnables.base.RunnableParallel

参考：

• Agents | 🦜️🔗 Langchain
• 概念解析：Concepts | 🦜️🔗 Langchain

agent 的一般逻辑：

• 负责把回答的问题通过 llm 分解成可是使用 tools 可是解决的问题
• 调用合适的 tool, 回答问题

• agent 本身可以利用 llm 自己的能力处理 tools 解决不了的问题

  • 判断问题是否解决
  • 一般性的问题回答，例如：问好，hello 等
  • 判断使用哪个工具
  • 判断工具回答结果
  • 判断接下来要使用哪个工具
• agent 能够结合多个 tool 类回答问题，例如：tool_1 的回答作为 tool_2 的输入

• 调用的工具 tools:

  • 知识库，eg: vector db
  • 搜索引擎
  • 程序，eg: python 函数等
  • 数据库

• 功能分解：

  Observation

  观察，上一步的执行结果获取

  Thought

  思考，针对上一步的结果，分析（1）问题解决了？（2）下一步使用什么工具

  Action

  行动，运行工具

  ?

  后续步骤： Observation, Thought, Action, …, Thought –> Answer

创建的 agent:

• 一般创建的 agent 是 AgentExecutor, 内部包含了 llm 、tools、 agent_object(Agent)
• 通过 AgentType(string 类型) 映射到 AGENT_TYPE 具体的类 类构建 agent_object

参考：

• LangChain干货（2）：ReAct框架，推理与行动的协同！ - 知乎
• REACT: Synergizing Reasoning and Acting In Language Models

名称由来：ReAct, [Re]sioning and [Act]ing, 推理和行动

这里主要记录，如何接入通过 server 暴露的模型访问

• ollama 接口
• lamacpp server 接口
• 其他 server api 接口

1
2
3
4
5
6
7
8

from llamaapi import LlamaAPI

# Replace 'Your_API_Token' with your actual API token
llama = LlamaAPI("Your_API_Token")

from langchain_experimental.llms import ChatLlamaAPI

model = ChatLlamaAPI(client=llama)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from langchain_openai import ChatOpenAI

inference_server_url = "http://localhost:8000/v1"

chat = ChatOpenAI(
    model="mosaicml/mpt-7b",
    openai_api_key="EMPTY",
    openai_api_base=inference_server_url,
    max_tokens=5,
    temperature=0,
)

1
2
3
4
5
6
7
8
9

from langchain_community.llms import VLLMOpenAI

llm = VLLMOpenAI(
    openai_api_key="EMPTY",
    openai_api_base="http://localhost:8000/v1",
    model_name="tiiuae/falcon-7b",
    model_kwargs={"stop": ["."]},
)
print(llm.invoke("Rome is"))

参考：

• 🦜🕸️LangGraph | 🦜️🔗 Langchain

概念：

• 入口 entrypoint:

  • 指定入口 Node 是哪一个
  • MessageGraph.set_entry_point("node_name")

• state

  • state 是 langgraph 的重要概念，node 会更新 state

  • state 的创建和传递

    • LangGraph 创建 state
    • 输入 message 会被放入到 state
    • state 被传入下一个执行的 node
    • node 执行后，output message 会放入到 state

  • state 的可能取值：

    • list

    • list[BaseMessage]

      • 消息列表，eg: [HumanMessage(), AIMessage()]

      • 使用例子：

        • 一个 langchain model 作为一个 MessageGraph 的 Node 可以正常处理传入的 state

          1 2 3 4

          def call_oracle(messages: list): return model.invoke(message) graph.add_node("oracle", call_oracle)

        • 注意： langchain Chain 接受的是 dict, 不是 list, 所以 不能作为 MessageGraph 的 Node

  • state 中的 额外信息

    • 通过 BaseMessage.additional_kwargs: dict, 传递

      • 比如： openai function call 功能：

        1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

        from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0) llm_with_tools = llm.bind_tools([Multiply]) llm_with_tools.invoke("what's 3 * 12") AIMessage( content="", additional_kwargs={ "tool_calls": [ { "id": "call_Q8ZQ97Qrj5zalugSkYMGV1Uo", "function": {"arguments": '{"a":3,"b":12}', "name": "Multiply"}, "type": "function", } ] }, )

        • 使用 additional_kwargs, 通过判断 tool_calls 中使用的 function -> name 来判断之前的 Node 推荐调用哪些 tool (function)

• node

  • node 有名称和能够把 state 作为输入的 callable

    • name: 名称

    • callable(state: list[BaseMessage])

      • eg: model

• router 和 conditional edges

  • conditional edge 包含内容：

    • router 函数： router_callable(state: list[BaseMessage]): router 判断上一步 state 应该被分发到哪个 node
    • 返回值和 Node 映射关系： return_value_2_node_name: dict: router 函数的返回值 和 MessageGraph 中的 Node 的名称映射

  • 使用逻辑： conditional edge 在执行完上一个 Node 后，调用 router

    1 2 3 4 5 6 7 8 9 10 11

    def router(state: List[BaseMessage]): tool_calls = state[-1].additional_kwargs.get("tool_calls", []) if len(tool_calls): return "multiply" else: return "end" graph.add_conditional_edges("oracle", router, { "multiply": "multiply", "end": END, })

    • 解释：

      • 这里的 router 判断 state 中最后一个 message 的信息，决定是调用 multply Node, 还是走到 END edge

1
2

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

参考：

• v2: How to add message history | 🦜️🔗 LangChain
• v1: {Beta} Memory | 🦜️🔗 LangChain

参考：

• How to return structured data from a model | 🦜️🔗 LangChain

使用 json, pydantic, yaml, … 格式化模型的输出结果，直接获得格式化对象。